# 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 struct from typing import Any, Dict, List, Optional from volatility.framework import interfaces, constants, exceptions from volatility.framework.configuration import requirements from volatility.framework.layers import physical, segmented, resources from volatility.framework.symbols import native class VmwareFormatException(exceptions.LayerException): """Thrown when an error occurs with the underlying VMware vmem file format.""" class VmwareLayer(segmented.SegmentedLayer): header_structure = "<4sII" group_structure = "64sQQ" def __init__(self, context: interfaces.context.ContextInterface, config_path: str, name: str, metadata: Optional[Dict[str, Any]] = None) -> None: # Construct these so we can use self.config self._context = context self._config_path = config_path self._page_size = 0x1000 self._base_layer, self._meta_layer = self.config["base_layer"], self.config["meta_layer"] # Then call the super, which will call load_segments (which needs the base_layer before it'll work) super().__init__(context, config_path = config_path, name = name, metadata = metadata) def _load_segments(self) -> None: """Loads up the segments from the meta_layer.""" self._read_header() def _read_header(self) -> None: """Checks the vmware header to make sure it's valid.""" if "vmware" not in self._context.symbol_space: self._context.symbol_space.append(native.NativeTable("vmware", native.std_ctypes)) meta_layer = self.context.layers.get(self._meta_layer, None) header_size = struct.calcsize(self.header_structure) data = meta_layer.read(0, header_size) magic, unknown, groupCount = struct.unpack(self.header_structure, data) if magic not in [b"\xD2\xBE\xD2\xBE"]: raise VmwareFormatException(self.name, "Wrong magic bytes for Vmware layer: {}".format(repr(magic))) # TODO: Change certain structure sizes based on the version version = magic[1] & 0xf group_size = struct.calcsize(self.group_structure) groups = {} for group in range(groupCount): name, tag_location, _unknown = struct.unpack( self.group_structure, meta_layer.read(header_size + (group * group_size), group_size)) name = name.rstrip(b"\x00") groups[name] = tag_location memory = groups[b"memory"] tags_read = False offset = memory tags = {} index_len = self._context.symbol_space.get_type("vmware!unsigned int").size while not tags_read: flags = ord(meta_layer.read(offset, 1)) name_len = ord(meta_layer.read(offset + 1, 1)) tags_read = (flags == 0) and (name_len == 0) if not tags_read: name = self._context.object("vmware!string", layer_name = self._meta_layer, offset = offset + 2, max_length = name_len) indicies_len = (flags >> 6) & 3 indicies = [] for index in range(indicies_len): indicies.append( self._context.object("vmware!unsigned int", offset = offset + name_len + 2 + (index * index_len), layer_name = self._meta_layer)) data = self._context.object("vmware!unsigned int", layer_name = self._meta_layer, offset = offset + 2 + name_len + (indicies_len * index_len)) tags[(name, tuple(indicies))] = (flags, data) offset += 2 + name_len + (indicies_len * index_len) + self._context.symbol_space.get_type("vmware!unsigned int").size if tags[("regionsCount", ())][1] == 0: raise VmwareFormatException(self.name, "VMware VMEM is not split into regions") for region in range(tags[("regionsCount", ())][1]): offset = tags[("regionPPN", (region, ))][1] * self._page_size mapped_offset = tags[("regionPageNum", (region, ))][1] * self._page_size length = tags[("regionSize", (region, ))][1] * self._page_size self._segments.append((offset, mapped_offset, length)) @property def dependencies(self) -> List[str]: return [self._base_layer, self._meta_layer] @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: """This vmware translation layer always requires a separate metadata layer.""" return [ requirements.TranslationLayerRequirement(name = 'base_layer', optional = False), requirements.TranslationLayerRequirement(name = 'meta_layer', optional = False) ] class VmwareStacker(interfaces.automagic.StackerLayerInterface): stack_order = 20 @classmethod def stack(cls, context: interfaces.context.ContextInterface, layer_name: str, progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]: """Attempt to stack this based on the starting information.""" memlayer = context.layers[layer_name] if not isinstance(memlayer, physical.FileLayer): return None location = memlayer.location if location.endswith(".vmem"): vmss = location[:-5] + ".vmss" vmsn = location[:-5] + ".vmsn" current_layer_name = context.layers.free_layer_name("VmwareMetaLayer") current_config_path = interfaces.configuration.path_join("automagic", "layer_stacker", "stack", current_layer_name) try: _ = resources.ResourceAccessor().open(vmss).read(10) context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmss context.layers.add_layer(physical.FileLayer(context, current_config_path, current_layer_name)) vmss_success = True except IOError: vmss_success = False if not vmss_success: try: _ = resources.ResourceAccessor().open(vmsn).read(10) context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmsn context.layers.add_layer(physical.FileLayer(context, current_config_path, current_layer_name)) vmsn_success = True except IOError: vmsn_success = False if not vmss_success and not vmsn_success: return None new_layer_name = context.layers.free_layer_name("VmwareLayer") context.config[interfaces.configuration.path_join(current_config_path, "base_layer")] = layer_name context.config[interfaces.configuration.path_join(current_config_path, "meta_layer")] = current_layer_name new_layer = VmwareLayer(context, current_config_path, new_layer_name) return new_layer return None