mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-11 20:27:38 +02:00
Merge pull request #1204 from volatilityfoundation/pedump_v2
Add pedump plugin and unified API for extracting PE files to disk
This commit is contained in:
ikelos
@@ -1,10 +1,9 @@
|
||||
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
import contextlib
|
||||
import datetime
|
||||
import logging
|
||||
import ntpath
|
||||
import re
|
||||
from typing import List, Optional, Type
|
||||
|
||||
@@ -14,7 +13,7 @@ from volatility3.framework.renderers import conversion, format_hints
|
||||
from volatility3.framework.symbols import intermed
|
||||
from volatility3.framework.symbols.windows.extensions import pe
|
||||
from volatility3.plugins import timeliner
|
||||
from volatility3.plugins.windows import info, pslist, psscan
|
||||
from volatility3.plugins.windows import info, pslist, psscan, pedump
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
@@ -23,7 +22,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
"""Lists the loaded modules in a particular windows memory image."""
|
||||
|
||||
_required_framework_version = (2, 0, 0)
|
||||
_version = (2, 0, 1)
|
||||
_version = (3, 0, 0)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
@@ -76,67 +75,11 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
default=False,
|
||||
optional=True,
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pedump", component=pedump.PEDump, version=(1, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def dump_pe(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
pe_table_name: str,
|
||||
dll_entry: interfaces.objects.ObjectInterface,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
layer_name: str = None,
|
||||
prefix: str = "",
|
||||
) -> Optional[interfaces.plugins.FileHandlerInterface]:
|
||||
"""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
|
||||
open_method: class for constructing output files
|
||||
|
||||
Returns:
|
||||
An open FileHandlerInterface object containing the complete data for the DLL or None in the case of failure
|
||||
"""
|
||||
try:
|
||||
try:
|
||||
name = dll_entry.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
name = "UnreadableDLLName"
|
||||
|
||||
if layer_name is None:
|
||||
layer_name = dll_entry.vol.layer_name
|
||||
|
||||
file_handle = open_method(
|
||||
"{}{}.{:#x}.{:#x}.dmp".format(
|
||||
prefix,
|
||||
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():
|
||||
file_handle.seek(offset)
|
||||
file_handle.write(data)
|
||||
except (
|
||||
IOError,
|
||||
exceptions.VolatilityException,
|
||||
OverflowError,
|
||||
ValueError,
|
||||
) as excp:
|
||||
vollog.debug(f"Unable to dump dll at offset {dll_entry.DllBase}: {excp}")
|
||||
return None
|
||||
return file_handle
|
||||
|
||||
def _generator(self, procs):
|
||||
pe_table_name = intermed.IntermediateSymbolTable.create(
|
||||
self.context, self.config_path, "windows", "pe", class_types=pe.class_types
|
||||
@@ -204,7 +147,7 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
|
||||
file_output = "Disabled"
|
||||
if self.config["dump"]:
|
||||
file_handle = self.dump_pe(
|
||||
file_output = pedump.PEDump.dump_ldr_entry(
|
||||
self.context,
|
||||
pe_table_name,
|
||||
entry,
|
||||
@@ -212,10 +155,10 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
proc_layer_name,
|
||||
prefix=f"pid.{proc_id}.",
|
||||
)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_handle.close()
|
||||
file_output = file_handle.preferred_filename
|
||||
|
||||
if not file_output:
|
||||
file_output = "Error outputting file"
|
||||
|
||||
try:
|
||||
dllbase = format_hints.Hex(entry.DllBase)
|
||||
except exceptions.InvalidAddressException:
|
||||
|
||||
@@ -6,7 +6,7 @@ from typing import Iterable
|
||||
|
||||
from volatility3.framework import interfaces
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.plugins.windows import poolscanner, dlllist, pslist, modules
|
||||
from volatility3.plugins.windows import poolscanner, modules, pedump
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
@@ -35,12 +35,6 @@ class ModScan(modules.Modules):
|
||||
requirements.VersionRequirement(
|
||||
name="modules", component=modules.Modules, version=(2, 0, 0)
|
||||
),
|
||||
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",
|
||||
description="Extract listed modules",
|
||||
@@ -58,6 +52,9 @@ class ModScan(modules.Modules):
|
||||
optional=True,
|
||||
default=None,
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pedump", component=pedump.PEDump, version=(1, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -9,7 +9,7 @@ from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.framework.symbols import intermed
|
||||
from volatility3.framework.symbols.windows.extensions import pe
|
||||
from volatility3.plugins.windows import pslist, dlllist
|
||||
from volatility3.plugins.windows import pslist, pedump
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
@@ -35,9 +35,6 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
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",
|
||||
description="Extract listed modules",
|
||||
@@ -55,6 +52,9 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
optional=True,
|
||||
default=None,
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pedump", component=pedump.PEDump, version=(1, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
def dump_module(self, session_layers, pe_table_name, mod):
|
||||
@@ -63,16 +63,15 @@ class Modules(interfaces.plugins.PluginInterface):
|
||||
)
|
||||
file_output = f"Cannot find a viable session layer for {mod.DllBase:#x}"
|
||||
if session_layer_name:
|
||||
file_handle = dlllist.DllList.dump_pe(
|
||||
file_output = pedump.PEDump.dump_ldr_entry(
|
||||
self.context,
|
||||
pe_table_name,
|
||||
mod,
|
||||
self.open,
|
||||
layer_name=session_layer_name,
|
||||
)
|
||||
file_output = "Error outputting file"
|
||||
if file_handle:
|
||||
file_output = file_handle.preferred_filename
|
||||
if not file_output:
|
||||
file_output = "Error outputting file"
|
||||
|
||||
return file_output
|
||||
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
import logging
|
||||
import ntpath
|
||||
from typing import List, Type, Optional
|
||||
|
||||
from volatility3.framework import constants, exceptions, interfaces, renderers
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.symbols import intermed
|
||||
from volatility3.framework.symbols.windows.extensions import pe
|
||||
from volatility3.plugins.windows import pslist, modules
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PEDump(interfaces.plugins.PluginInterface):
|
||||
"""Allows extracting PE Files from a specific address in a specific address space"""
|
||||
|
||||
_required_framework_version = (2, 0, 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.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(2, 0, 0)
|
||||
),
|
||||
requirements.ListRequirement(
|
||||
name="pid",
|
||||
element_type=int,
|
||||
description="Process IDs to include (all other processes are excluded)",
|
||||
optional=True,
|
||||
),
|
||||
requirements.IntRequirement(
|
||||
name="base",
|
||||
description="Base address to reconstruct a PE file",
|
||||
optional=False,
|
||||
),
|
||||
requirements.BooleanRequirement(
|
||||
name="kernel_module",
|
||||
description="Extract from kernel address space.",
|
||||
default=False,
|
||||
optional=True,
|
||||
),
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def dump_pe(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
pe_table_name: str,
|
||||
layer_name: str,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
file_name: str,
|
||||
base: int,
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Returns the filename of the dump file or None
|
||||
"""
|
||||
try:
|
||||
file_handle = open_method(file_name)
|
||||
|
||||
dos_header = context.object(
|
||||
pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER",
|
||||
offset=base,
|
||||
layer_name=layer_name,
|
||||
)
|
||||
|
||||
for offset, data in dos_header.reconstruct():
|
||||
file_handle.seek(offset)
|
||||
file_handle.write(data)
|
||||
except (
|
||||
IOError,
|
||||
exceptions.VolatilityException,
|
||||
OverflowError,
|
||||
ValueError,
|
||||
) as excp:
|
||||
vollog.debug(f"Unable to dump PE file at offset {base}: {excp}")
|
||||
return None
|
||||
finally:
|
||||
file_handle.close()
|
||||
|
||||
return file_handle.preferred_filename
|
||||
|
||||
@classmethod
|
||||
def dump_ldr_entry(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
pe_table_name: str,
|
||||
ldr_entry: interfaces.objects.ObjectInterface,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
layer_name: str = None,
|
||||
prefix: str = "",
|
||||
) -> Optional[str]:
|
||||
"""Extracts the PE file referenced an LDR_DATA_TABLE_ENTRY (DLL, kernel module) instance
|
||||
|
||||
Args:
|
||||
context: the context to operate upon
|
||||
pe_table_name: the name for the symbol table containing the PE format symbols
|
||||
ldr_entry: the object representing the module
|
||||
open_method: class for constructing output files
|
||||
layer_name: the layer that the DLL lives within
|
||||
prefix: optional string to prepend to filename
|
||||
Returns:
|
||||
The output file name or None in the case of failure
|
||||
"""
|
||||
try:
|
||||
name = ldr_entry.FullDllName.get_string()
|
||||
except exceptions.InvalidAddressException:
|
||||
name = "UnreadableDLLName"
|
||||
|
||||
if layer_name is None:
|
||||
layer_name = ldr_entry.vol.layer_name
|
||||
|
||||
file_name = "{}{}.{:#x}.{:#x}.dmp".format(
|
||||
prefix,
|
||||
ntpath.basename(name),
|
||||
ldr_entry.vol.offset,
|
||||
ldr_entry.DllBase,
|
||||
)
|
||||
|
||||
return cls.dump_pe(
|
||||
context,
|
||||
pe_table_name,
|
||||
layer_name,
|
||||
open_method,
|
||||
file_name,
|
||||
ldr_entry.DllBase,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def dump_pe_at_base(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
pe_table_name: str,
|
||||
layer_name: str,
|
||||
open_method: Type[interfaces.plugins.FileHandlerInterface],
|
||||
proc_offset: int,
|
||||
pid: int,
|
||||
base: int,
|
||||
) -> Optional[str]:
|
||||
file_name = "PE.{:#x}.{:d}.{:#x}.dmp".format(
|
||||
proc_offset,
|
||||
pid,
|
||||
base,
|
||||
)
|
||||
|
||||
return PEDump.dump_pe(
|
||||
context, pe_table_name, layer_name, open_method, file_name, base
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def dump_kernel_pe_at_base(cls, context, kernel, pe_table_name, open_method, base):
|
||||
session_layers = modules.Modules.get_session_layers(
|
||||
context, kernel.layer_name, kernel.symbol_table_name
|
||||
)
|
||||
|
||||
session_layer_name = modules.Modules.find_session_layer(
|
||||
context, session_layers, base
|
||||
)
|
||||
|
||||
if session_layer_name:
|
||||
system_pid = 4
|
||||
|
||||
file_output = PEDump.dump_pe_at_base(
|
||||
context,
|
||||
pe_table_name,
|
||||
session_layer_name,
|
||||
open_method,
|
||||
0,
|
||||
system_pid,
|
||||
base,
|
||||
)
|
||||
|
||||
if file_output:
|
||||
yield system_pid, "Kernel", file_output
|
||||
else:
|
||||
vollog.warning(
|
||||
"Unable to find a session layer with the provided base address mapped in the kernel."
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def dump_processes(
|
||||
cls, context, kernel, pe_table_name, open_method, filter_func, base
|
||||
):
|
||||
""" """
|
||||
|
||||
for proc in pslist.PsList.list_processes(
|
||||
context=context,
|
||||
layer_name=kernel.layer_name,
|
||||
symbol_table=kernel.symbol_table_name,
|
||||
filter_func=filter_func,
|
||||
):
|
||||
pid = proc.UniqueProcessId
|
||||
proc_name = proc.ImageFileName.cast(
|
||||
"string",
|
||||
max_length=proc.ImageFileName.vol.count,
|
||||
errors="replace",
|
||||
)
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
|
||||
file_output = PEDump.dump_pe_at_base(
|
||||
context,
|
||||
pe_table_name,
|
||||
proc_layer_name,
|
||||
open_method,
|
||||
proc.vol.offset,
|
||||
pid,
|
||||
base,
|
||||
)
|
||||
|
||||
if file_output:
|
||||
yield pid, proc_name, file_output
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
if self.config["kernel_module"] and self.config["pid"]:
|
||||
vollog.error("Only --kernel_module or --pid should be set. Not both")
|
||||
return
|
||||
|
||||
if not self.config["kernel_module"] and not self.config["pid"]:
|
||||
vollog.error("--kernel_module or --pid must be set")
|
||||
return
|
||||
|
||||
if self.config["kernel_module"]:
|
||||
pe_files = self.dump_kernel_pe_at_base(
|
||||
self.context, kernel, pe_table_name, self.open, self.config["base"]
|
||||
)
|
||||
else:
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
|
||||
pe_files = self.dump_processes(
|
||||
self.context,
|
||||
kernel,
|
||||
pe_table_name,
|
||||
self.open,
|
||||
filter_func,
|
||||
self.config["base"],
|
||||
)
|
||||
|
||||
for pid, proc_name, file_output in pe_files:
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
pid,
|
||||
proc_name,
|
||||
file_output,
|
||||
),
|
||||
)
|
||||
|
||||
def run(self):
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
("PID", int),
|
||||
("Process", str),
|
||||
("File output", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
Reference in New Issue
Block a user