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

Dump ELFs to file

Browse Source
This commit is contained in:
cstation
2023-01-13 12:27:31 +01:00
parent 1641a6e4c4
commit 558b31cbdc
2 changed files with 180 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/linux/elfs.py
+111 -2
View File
@@ -4,20 +4,26 @@
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
from typing import List
import logging
from typing import List, Optional, Type
from volatility3.framework import renderers, interfaces
from volatility3.framework import constants, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.linux.extensions import elf
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
class Elfs(plugins.PluginInterface):
"""Lists all memory mapped ELF files for all processes."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -36,9 +42,95 @@ class Elfs(plugins.PluginInterface):
element_type=int,
optional=True,
),
requirements.BooleanRequirement(
name="dump",
description="Extract listed processes",
default=False,
optional=True,
),
]
@classmethod
def elf_dump(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
elf_table_name: str,
vma: interfaces.objects.ObjectInterface,
task: interfaces.objects.ObjectInterface,
open_method: Type[interfaces.plugins.FileHandlerInterface],
) -> Optional[interfaces.plugins.FileHandlerInterface]:
"""Extracts an ELF as a FileHandlerInterface
Args:
context: the context to operate upon
layer_name: The name of the layer on which to operate
elf_table_name: the name for the symbol table containing the symbols for ELF-files
vma: virtual memory allocation of ELF
task: the task object whose memory should be output
open_method: class to provide context manager for opening the file
Returns:
An open FileHandlerInterface object containing the complete data for the task or None in the case of failure
"""
proc_layer = context.layers[layer_name]
file_handle = None
try:
elf_object = context.object(
elf_table_name + constants.BANG + "Elf",
offset=vma.vm_start,
layer_name=layer_name,
)
if not elf_object.is_valid():
return None
sections = {}
# TODO: Apply more effort to reconstruct ELF, e.g.: https://github.com/enbarberis/core2ELF64 ?
for phdr in elf_object.get_program_headers():
if phdr.p_type != 1: # PT_LOAD = 1
continue
start = phdr.p_vaddr
size = phdr.p_memsz
end = start + size
# Use complete memory pages for dumping
# If start isn't a multiple of 4096, stick to the highest multiple < start
# If end isn't a multiple of 4096, stick to the lowest multiple > end
if start % 4096:
start = start & ~0xFFF
if end % 4096:
end = (end & ~0xFFF) + 4096
real_size = end - start
if real_size < 0 or real_size > 100000000:
continue
sections[start] = real_size
elf_data = b""
for section_start in sorted(sections.keys()):
read_size = sections[section_start]
buf = proc_layer.read(vma.vm_start + section_start, read_size, pad=True)
elf_data = elf_data + buf
file_handle = open_method(
f"pid.{task.pid}.{utility.array_to_string(task.comm)}.{vma.vm_start:#x}.dmp"
)
file_handle.write(elf_data)
except Exception as e:
vollog.debug(f"Unable to dump ELF with pid {task.pid}: {e}")
return file_handle
def _generator(self, tasks):
elf_table_name = intermed.IntermediateSymbolTable.create(
self.context, self.config_path, "linux", "elf", class_types=elf.class_types
)
for task in tasks:
proc_layer_name = task.add_process_layer()
if not proc_layer_name:
@@ -60,6 +152,21 @@ class Elfs(plugins.PluginInterface):
path = vma.get_name(self.context, task)
file_output = "Disabled"
if self.config["dump"]:
file_handle = self.elf_dump(
self.context,
proc_layer_name,
elf_table_name,
vma,
task,
self.open,
)
file_output = "Error outputting file"
if file_handle:
file_handle.close()
file_output = str(file_handle.preferred_filename)
yield (
0,
(
@@ -68,6 +175,7 @@ class Elfs(plugins.PluginInterface):
format_hints.Hex(vma.vm_start),
format_hints.Hex(vma.vm_end),
path,
file_output,
),
)
@@ -81,6 +189,7 @@ class Elfs(plugins.PluginInterface):
("Start", format_hints.Hex),
("End", format_hints.Hex),
("File Path", str),
("File Output", str),
],
self._generator(
pslist.PsList.list_tasks(
volatility3/framework/plugins/linux/pslist.py
+69 -32
View File
@@ -1,12 +1,15 @@
# This file is Copyright 2021 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 Callable, Iterable, List, Any, Tuple
from typing import Any, Callable, Iterable, List
from volatility3.framework import renderers, interfaces
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import intermed
from volatility3.framework.symbols.linux.extensions import elf
from volatility3.plugins.linux import elfs
class PsList(interfaces.plugins.PluginInterface):
@@ -24,6 +27,9 @@ class PsList(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="elfs", plugin=elfs.Elfs, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
description="Filter on specific process IDs",
@@ -42,6 +48,12 @@ class PsList(interfaces.plugins.PluginInterface):
optional=True,
default=False,
),
requirements.BooleanRequirement(
name="dump",
description="Extract listed processes",
optional=True,
default=False,
),
]
@classmethod
@@ -66,38 +78,12 @@ class PsList(interfaces.plugins.PluginInterface):
else:
return lambda _: False
def _get_task_fields(
self, task: interfaces.objects.ObjectInterface, decorate_comm: bool = False
) -> Tuple[int, int, int, str]:
"""Extract the fields needed for the final output
Args:
task: A task object from where to get the fields.
decorate_comm: If True, it decorates the comm string of
- User threads: in curly brackets,
- Kernel threads: in square brackets
Defaults to False.
Returns:
A tuple with the fields to show in the plugin output.
"""
pid = task.tgid
tid = task.pid
ppid = task.parent.tgid if task.parent else 0
name = utility.array_to_string(task.comm)
if decorate_comm:
if task.is_kernel_thread:
name = f"[{name}]"
elif task.is_user_thread:
name = f"{{{name}}}"
task_fields = (format_hints.Hex(task.vol.offset), pid, tid, ppid, name)
return task_fields
def _generator(
self,
pid_filter: Callable[[Any], bool],
include_threads: bool = False,
decorate_comm: bool = False,
dump: bool = False,
):
"""Generates the tasks list.
@@ -110,14 +96,63 @@ class PsList(interfaces.plugins.PluginInterface):
- User threads: in curly brackets,
- Kernel threads: in square brackets
Defaults to False.
dump: If True, the main executable of the process is written to a file
Defaults to False.
Yields:
Each rows
"""
for task in self.list_tasks(
self.context, self.config["kernel"], pid_filter, include_threads
):
row = self._get_task_fields(task, decorate_comm)
yield (0, row)
elf_table_name = intermed.IntermediateSymbolTable.create(
self.context,
self.config_path,
"linux",
"elf",
class_types=elf.class_types,
)
file_output = "Disabled"
if dump:
proc_layer_name = task.add_process_layer()
if not proc_layer_name:
continue
# Find the vma that belongs to the main ELF of the process
file_output = "Error outputting file"
for v in task.mm.get_mmap_iter():
if v.vm_start == task.mm.start_code:
file_handle = elfs.Elfs.elf_dump(
self.context,
proc_layer_name,
elf_table_name,
v,
task,
self.open,
)
if file_handle:
file_output = str(file_handle.preferred_filename)
file_handle.close()
break
pid = task.tgid
tid = task.pid
ppid = task.parent.tgid if task.parent else 0
name = utility.array_to_string(task.comm)
if decorate_comm:
if task.is_kernel_thread:
name = f"[{name}]"
elif task.is_user_thread:
name = f"{{{name}}}"
yield 0, (
format_hints.Hex(task.vol.offset),
pid,
tid,
ppid,
name,
file_output,
)
@classmethod
def list_tasks(
@@ -155,6 +190,7 @@ class PsList(interfaces.plugins.PluginInterface):
pids = self.config.get("pid")
include_threads = self.config.get("threads")
decorate_comm = self.config.get("decorate_comm")
dump = self.config.get("dump")
filter_func = self.create_pid_filter(pids)
columns = [
@@ -163,7 +199,8 @@ class PsList(interfaces.plugins.PluginInterface):
("TID", int),
("PPID", int),
("COMM", str),
("File output", str),
]
return renderers.TreeGrid(
columns, self._generator(filter_func, include_threads, decorate_comm)
columns, self._generator(filter_func, include_threads, decorate_comm, dump)
)