From dfe635e20b1ad7cc5adaba45d4076625c0051d9c Mon Sep 17 00:00:00 2001 From: Melissa Eckardt Date: Tue, 5 May 2026 16:50:15 +0200 Subject: [PATCH] Add ElfCoreDump plugins --- .../framework/plugins/linux/elfcoredump.py | 161 +++++++++++ .../framework/plugins/windows/elfcoredump.py | 165 +++++++++++ .../symbols/linux/utilities/coredumpwriter.py | 262 ++++++++++++++++++ 3 files changed, 588 insertions(+) create mode 100644 volatility3/framework/plugins/linux/elfcoredump.py create mode 100644 volatility3/framework/plugins/windows/elfcoredump.py create mode 100644 volatility3/framework/symbols/linux/utilities/coredumpwriter.py diff --git a/volatility3/framework/plugins/linux/elfcoredump.py b/volatility3/framework/plugins/linux/elfcoredump.py new file mode 100644 index 0000000000..ee9121f976 --- /dev/null +++ b/volatility3/framework/plugins/linux/elfcoredump.py @@ -0,0 +1,161 @@ +# This file is Copyright 2026 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 functools import partial +from typing import BinaryIO, List, Generator, Tuple + +from volatility3.framework import renderers, interfaces +from volatility3.framework.configuration import requirements +from volatility3.framework.interfaces.layers import DataLayerInterface +from volatility3.framework.layers import intel +from volatility3.framework.objects import utility +from volatility3.framework.symbols.linux.utilities.coredumpwriter import CoreDumpWriter, PF_R, PF_W, PF_X +from volatility3.plugins.linux import proc, pslist + +vollog = logging.getLogger(__name__) + + +class ElfCoreDump(interfaces.plugins.PluginInterface): + """Creates a process core dump in ELF format containing the address space + and segment names. Thread state is not included.""" + + _required_framework_version = (2, 4, 0) + _version = (1, 0, 0) + + MAX_VMA_SIZE = 1024 * 1024 * 1024 # Don't dump VMAs above 1 GB + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + @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="Linux kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.IntRequirement( + name="pid", + description="Process ID to dump", + optional=False, + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(4, 0, 0) + ), + requirements.VersionRequirement( + name="proc", component=proc.Maps, version=(1, 0, 3) + ), + ] + + @staticmethod + def prot_to_flags(prot: str) -> int: + flags = 0 + if "r" in prot: + flags |= PF_R + if "w" in prot: + flags |= PF_W + if "x" in prot: + flags |= PF_X + + return flags + + def write_segment_callback( + self, + proc_layer: DataLayerInterface, + file_handle: BinaryIO, + seg_start: int, + seg_size: int + ) -> None: + chunk_size = 1024 * 1024 * 10 + offset = seg_start + while offset < seg_start + seg_size: + to_read = min(chunk_size, seg_start + seg_size - offset) + data = proc_layer.read(offset, to_read, pad=True) + if not data: + break + file_handle.write(data) + offset += to_read + + def dump_process(self, task: interfaces.objects.ObjectInterface) -> str: + parent_layer = self.context.layers[task.vol.layer_name] + if not isinstance(parent_layer, intel.Intel): + raise TypeError("Process parent layer is not a translation layer, unable to get bitness") + + writer = CoreDumpWriter(parent_layer.bits_per_register) + for vma in proc.Maps.list_vmas(task): + if not vma.vm_start or not vma.vm_end: + continue + if vma.vm_end - vma.vm_start > self.MAX_VMA_SIZE: + vollog.debug( + f"VMA at 0x{vma.vm_start:x} over sanity-check size, not dumping" + ) + continue + + vma_name = vma.get_name(self.context, task) or "" + if not vma_name.startswith("/"): + vma_name = "" # only include real file mappings in the FILE note + + writer.add_segment(vma.vm_start, + vma.vm_end - vma.vm_start, + self.prot_to_flags(vma.get_protection()), + vma_name) + + proc_id = task.pid + proc_layer_name = task.add_process_layer() + proc_layer = self.context.layers[proc_layer_name] + + file = f"core.{proc_id}.elf" + with open(file, "wb") as fp: + writer.dump(fp, partial(self.write_segment_callback, proc_layer)) + + return file + + def _generator( + self, task: interfaces.objects.ObjectInterface + ) -> Generator[ + Tuple[ + int, + Tuple[ + int, + str, + str, + ], + ], + None, + None, + ]: + process_name = utility.array_to_string(task.comm) + + file = self.dump_process(task) + + yield ( + 0, + ( + task.pid, + process_name, + file, + ), + ) + + def run(self) -> renderers.TreeGrid: + filter_func = pslist.PsList.create_pid_filter([self.config.get("pid")]) + procs = list(pslist.PsList.list_tasks( + self.context, + self.config["kernel"], + filter_func, + )) + if len(procs) != 1: + raise Exception("Process not found") + + return renderers.TreeGrid( + [ + ("PID", int), + ("Process", str), + ("File", str), + ], + self._generator(procs[0]), + ) diff --git a/volatility3/framework/plugins/windows/elfcoredump.py b/volatility3/framework/plugins/windows/elfcoredump.py new file mode 100644 index 0000000000..d1fd006c41 --- /dev/null +++ b/volatility3/framework/plugins/windows/elfcoredump.py @@ -0,0 +1,165 @@ +# This file is Copyright 2026 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 functools import partial +from typing import BinaryIO, List, Generator, Tuple + +from volatility3.framework import renderers, interfaces, exceptions +from volatility3.framework.configuration import requirements +from volatility3.framework.interfaces.layers import DataLayerInterface +from volatility3.framework.layers import intel +from volatility3.framework.objects import utility +from volatility3.framework.symbols.linux.utilities.coredumpwriter import CoreDumpWriter, PF_R, PF_W, PF_X +from volatility3.plugins.windows import pslist, vadinfo + +vollog = logging.getLogger(__name__) + + +class ElfCoreDump(interfaces.plugins.PluginInterface): + """Creates a process core dump in ELF format containing the address space + and segment names. Thread state is not included.""" + + _required_framework_version = (2, 4, 0) + _version = (1, 0, 0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + @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.IntRequirement( + name="pid", + description="Process ID to dump", + optional=False, + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(3, 0, 0) + ), + requirements.VersionRequirement( + name="vadinfo", component=vadinfo.VadInfo, version=(2, 0, 1) + ), + ] + + @staticmethod + def prot_to_flags(prot: str) -> int: + flags = 0 + if "READ" in prot or "WRITECOPY" in prot: + flags |= PF_R + if "WRITE" in prot or "WRITECOPY" in prot: + flags |= PF_W + if "EXECUTE" in prot: + flags |= PF_X + + return flags + + def write_segment_callback( + self, + proc_layer: DataLayerInterface, + file_handle: BinaryIO, + seg_start: int, + seg_size: int + ) -> None: + chunk_size = 1024 * 1024 * 10 + offset = seg_start + while offset < seg_start + seg_size: + to_read = min(chunk_size, seg_start + seg_size - offset) + data = proc_layer.read(offset, to_read, pad=True) + if not data: + break + file_handle.write(data) + offset += to_read + + def dump_process(self, proc: interfaces.objects.ObjectInterface) -> str: + kernel = self.context.modules[self.config["kernel"]] + + parent_layer = self.context.layers[proc.vol.layer_name] + if not isinstance(parent_layer, intel.Intel): + raise TypeError("Process parent layer is not a translation layer, unable to get bitness") + + writer = CoreDumpWriter(parent_layer.bits_per_register) + for vad in vadinfo.VadInfo.list_vads(proc): + if vad.get_commit_charge() == 0 and \ + isinstance(vad.get_file_name(), renderers.NotApplicableValue): + continue + + writer.add_segment(vad.get_start(), + vad.get_size(), + self.prot_to_flags(vad.get_protection( + vadinfo.VadInfo.protect_values( + self.context, + kernel.layer_name, + kernel.symbol_table_name, + ), + vadinfo.winnt_protections, + )), + vad.get_file_name()) + + try: + proc_id = proc.UniqueProcessId + proc_layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException as excp: + raise Exception(f"Process {proc_id}: invalid address {excp.invalid_address}" + f" in layer {excp.layer_name}") from excp + + proc_layer = self.context.layers[proc_layer_name] + + file = f"core.{proc_id}.elf" + with open(file, "wb") as fp: + writer.dump(fp, partial(self.write_segment_callback, proc_layer)) + + return file + + def _generator( + self, proc: interfaces.objects.ObjectInterface + ) -> Generator[ + Tuple[ + int, + Tuple[ + int, + str, + str, + ], + ], + None, + None, + ]: + process_name = utility.array_to_string(proc.ImageFileName) + + file = self.dump_process(proc) + + yield ( + 0, + ( + proc.UniqueProcessId, + process_name, + file, + ), + ) + + def run(self) -> renderers.TreeGrid: + filter_func = pslist.PsList.create_pid_filter([self.config.get("pid")]) + procs = list(pslist.PsList.list_processes( + context=self.context, + kernel_module_name=self.config["kernel"], + filter_func=filter_func, + )) + if len(procs) != 1: + raise Exception("Process not found") + + return renderers.TreeGrid( + [ + ("PID", int), + ("Process", str), + ("File", str), + ], + self._generator(procs[0]), + ) diff --git a/volatility3/framework/symbols/linux/utilities/coredumpwriter.py b/volatility3/framework/symbols/linux/utilities/coredumpwriter.py new file mode 100644 index 0000000000..0dc291dbde --- /dev/null +++ b/volatility3/framework/symbols/linux/utilities/coredumpwriter.py @@ -0,0 +1,262 @@ +# This file is Copyright 2026 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 struct +from dataclasses import dataclass +from typing import BinaryIO, Callable, Union + +from volatility3.framework.renderers import NotApplicableValue + +vollog = logging.getLogger(__name__) + + +NT_FILE = 0x46494C45 # 'FILE' in big endian + +# Alignment values +NOTE_ALIGN = 4 +PAGE_SIZE = 0x1000 + +# Program header flags +PF_X = 1 +PF_W = 2 +PF_R = 4 + +# Program header types +PT_LOAD = 1 +PT_NOTE = 4 + + +def align(x, a): + return (x + (a - 1)) & ~(a - 1) + + +@dataclass +class Segment: + vaddr: int + """Virtual address of the segment.""" + size: int + """Size of the segment. On-disk and in-mem size is identical for dumps.""" + flags: int + """Segment permissions (combination of PF_* values).""" + name: str + """Name of the segment (i.e. image path), can be empty.""" + offset: int = 0 + """Internal field for serialization purposes.""" + + +class CoreDumpWriter: + """Creates an ELF core dump file for a process.""" + + def __init__(self, bits: int) -> None: + self.segments: list[Segment] = [] + self.bits = bits + + if self.bits == 32: + self.machine = 3 # EM_X86_86 + self.program_header_size = 32 + else: + self.machine = 0x3E # EM_X86_64 + self.program_header_size = 56 + + def add_segment( + self, vaddr: int, size: int, flags: int, name: Union[str, NotApplicableValue] + ) -> None: + """Adds information about a virtual memory segment to the dump.""" + if isinstance(name, NotApplicableValue): + name = "" + self.segments.append(Segment(vaddr, size, flags, name)) + + def build_note(self, name: str, n_type: int, desc: bytes) -> bytes: + """ + Builds a Note Section with the given specifications. + + Spec: https://refspecs.linuxfoundation.org/elf/gabi4+/ch5.pheader.html#note_section + """ + namesz = len(name) + 1 + descsz = len(desc) + + out = struct.pack(" bytes: + """Builds a CORE note of type NT_FILE, containing all of our segment names.""" + file_entries = [] + filenames = b"" + + for seg in self.segments: + if seg.name: + start = seg.vaddr + end = start + seg.size + + file_entries.append((start, end)) + filenames += seg.name.encode("utf-8") + b"\x00" + + fmt = "I" if self.bits == 32 else "Q" + note_desc = struct.pack(f"<{fmt}{fmt}", len(file_entries), PAGE_SIZE) + + for s, e in file_entries: + note_desc += struct.pack(f"<{fmt}{fmt}{fmt}", s, e, 0) # third word is file_ofs + + note_desc += filenames + + return self.build_note("CORE", NT_FILE, note_desc) + + def _make_elf_header(self, phnum: int) -> bytes: + """ + Creates an ELF header of type ET_CORE. + + Spec: https://refspecs.linuxfoundation.org/elf/gabi4+/ch4.eheader.html + """ + if phnum >= 0xFFFF: + # A file with this many segments is not impossible, but it would require + # extra hoops (putting the count into the first section). + raise ValueError(f"Unsupported number of segments for dump file: {phnum}") + + if self.bits == 32: + fmt = " bytes: + """ + Serializes a program header. + + Spec: https://refspecs.linuxfoundation.org/elf/gabi4+/ch5.pheader.html + """ + if self.bits == 32: + return struct.pack( + " None: + """ + Writes the core dump into the given BinaryIO. data_callback will be called with the + file handle, start address and size whenever actual segment data needs to be written. + """ + note_blobs = [self.build_file_note()] + + phnum = len(note_blobs) + len(self.segments) + + elf_header = self._make_elf_header(phnum) + data_start = align(len(elf_header) + phnum * self.program_header_size, + PAGE_SIZE) + + offset = data_start + phdr_blob = b"" + + for note_blob in note_blobs: + phdr_blob += self._make_phdr(PT_NOTE, 0, offset, + 0, len(note_blob), NOTE_ALIGN) + offset += align(len(note_blob), NOTE_ALIGN) + + for seg in self.segments: + offset = align(offset, PAGE_SIZE) + + phdr_blob += self._make_phdr(PT_LOAD, seg.flags, offset, + seg.vaddr, seg.size, PAGE_SIZE) + + seg.offset = offset + offset += seg.size + + if fp.tell() != 0: + raise ValueError("File passed to CoreDumpWriter.dump() must be empty") + + fp.write(elf_header) + fp.write(phdr_blob) + + # pad to data_start + cur = fp.tell() + fp.write(b"\x00" * (data_start - cur)) + + # write notes + for note_blob in note_blobs: + fp.write(note_blob) + fp.write(b"\x00" * (align(len(note_blob), NOTE_ALIGN) - len(note_blob))) + + # write segments + for seg in self.segments: + cur = fp.tell() + if cur < seg.offset: + fp.write(b"\x00" * (seg.offset - cur)) + + data_callback(fp, seg.vaddr, seg.size)