mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-22 14:32:21 +02:00
Mike Auty
ea629591ef
This highlights that the FileHandler class can also be seen as a method similar to open, and it removes unnecessary context managers, allowing plugins to close files as they wish (they must, however, remember to close the file for it to be committed).
99 lines
4.5 KiB
Python
99 lines
4.5 KiB
Python
# 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 os
|
|
from typing import List, Optional, Type
|
|
|
|
from volatility.framework import renderers, interfaces, constants, exceptions
|
|
from volatility.framework.configuration import requirements
|
|
from volatility.framework.interfaces import plugins
|
|
|
|
vollog = logging.getLogger(__name__)
|
|
|
|
|
|
class LayerWriter(plugins.PluginInterface):
|
|
"""Runs the automagics and writes out the primary layer produced by the stacker."""
|
|
|
|
default_output_name = "output.raw"
|
|
default_block_size = 0x500000
|
|
|
|
_required_framework_version = (2, 0, 0)
|
|
_version = (2, 0, 0)
|
|
|
|
@classmethod
|
|
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
|
return [
|
|
requirements.TranslationLayerRequirement(name = 'primary',
|
|
description = 'Memory layer for the kernel',
|
|
architectures = ["Intel32", "Intel64"]),
|
|
requirements.StringRequirement(name = 'output',
|
|
description = 'Filename to output the chosen layer',
|
|
optional = True,
|
|
default = cls.default_output_name),
|
|
requirements.IntRequirement(name = 'block_size',
|
|
description = "Size of blocks to copy over",
|
|
default = cls.default_block_size,
|
|
optional = True)
|
|
]
|
|
|
|
@classmethod
|
|
def write_layer(
|
|
cls,
|
|
context: interfaces.context.ContextInterface,
|
|
layer_name: str,
|
|
preferred_name: str,
|
|
open_method: Type[plugins.FileHandlerInterface],
|
|
chunk_size: Optional[int] = None,
|
|
progress_callback: Optional[constants.ProgressCallback] = None) -> Optional[plugins.FileHandlerInterface]:
|
|
"""Produces a FileHandler from the named layer in the provided context or None on failure
|
|
|
|
Args:
|
|
context: the context from which to read the memory layer
|
|
layer_name: the name of the layer to write out
|
|
preferred_name: a string with the preferred filename for hte file
|
|
chunk_size: an optional size for the chunks that should be written (defaults to 0x500000)
|
|
open_method: class for creating FileHandler context managers
|
|
progress_callback: an optional function that takes a percentage and a string that displays output
|
|
"""
|
|
|
|
if layer_name not in context.layers:
|
|
raise exceptions.LayerException("Layer not found")
|
|
layer = context.layers[layer_name]
|
|
|
|
if chunk_size is None:
|
|
chunk_size = cls.default_block_size
|
|
|
|
file_handle = open_method(preferred_name)
|
|
for i in range(0, layer.maximum_address, chunk_size):
|
|
current_chunk_size = min(chunk_size, layer.maximum_address - i)
|
|
data = layer.read(i, current_chunk_size, pad = True)
|
|
file_handle.write(data)
|
|
if progress_callback:
|
|
progress_callback((i / layer.maximum_address) * 100, 'Writing layer {}'.format(layer_name))
|
|
return file_handle
|
|
|
|
def _generator(self):
|
|
if self.config['primary'] not in self.context.layers:
|
|
yield 0, ('Layer Name does not exist',)
|
|
elif os.path.exists(self.config.get('output', self.default_output_name)):
|
|
yield 0, ('Refusing to overwrite existing output file',)
|
|
else:
|
|
output_name = self.config.get('output', self.default_output_name)
|
|
try:
|
|
file_handle = self.write_layer(self.context,
|
|
self.config['primary'],
|
|
output_name,
|
|
self.open,
|
|
self.config.get('block_size', self.default_block_size),
|
|
progress_callback = self._progress_callback)
|
|
file_handle.close()
|
|
except IOError as excp:
|
|
yield 0, ('Layer cannot be written to {}: {}'.format(self.config['output_name'], excp),)
|
|
|
|
yield 0, ('Layer has been written to {}'.format(output_name),)
|
|
|
|
def run(self):
|
|
return renderers.TreeGrid([("Status", str)], self._generator())
|