diff --git a/volatility/framework/automagic/__init__.py b/volatility/framework/automagic/__init__.py index 56f7fa63d..22ba74445 100644 --- a/volatility/framework/automagic/__init__.py +++ b/volatility/framework/automagic/__init__.py @@ -29,6 +29,11 @@ linux_automagic = ['ConstructionMagic', 'LinuxSymbolCache', 'LinuxSymbolFinder'] +mac_automagic = ['ConstructionMagic', + 'LayerStacker', + 'MacSymbolCache', + 'MacSymbolFinder'] + def available(context: interfaces.context.ContextInterface) \ -> typing.List[interfaces.automagic.AutomagicInterface]: @@ -59,6 +64,9 @@ def choose_automagic(automagics, plugin): elif plugin_category == 'linux': if amagic.__class__.__name__ in linux_automagic: output += [amagic] + elif plugin_category == 'mac': + if amagic.__class__.__name__ in mac_automagic: + output += [amagic] else: return automagics vollog.info("Restricting automagics to: {}".format([x.__class__.__name__ for x in output])) diff --git a/volatility/framework/automagic/linux.py b/volatility/framework/automagic/linux.py index 5f6e7e6a4..013932c6a 100644 --- a/volatility/framework/automagic/linux.py +++ b/volatility/framework/automagic/linux.py @@ -4,7 +4,7 @@ import typing import volatility.framework.objects.utility from volatility.framework import interfaces, constants, validity, exceptions, layers from volatility.framework import symbols, objects -from volatility.framework.automagic import linux_symbol_cache +from volatility.framework.automagic import symbol_cache from volatility.framework.configuration import requirements from volatility.framework.layers import intel, scanners from volatility.framework.symbols import linux @@ -21,13 +21,13 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface): config_path: str) -> None: super().__init__(context, config_path) self._requirements = [] # type: typing.List[typing.Tuple[str, interfaces.configuration.ConstructableRequirementInterface]] - self._linux_banners_ = {} # type: linux_symbol_cache.LinuxBanners + self._linux_banners_ = {} # type: symbol_cache.BannersType @property - def _linux_banners(self) -> linux_symbol_cache.LinuxBanners: + def _linux_banners(self) -> symbol_cache.BannersType: """Creates a cached copy of the results, but only it's been requested""" if not self._linux_banners_: - self._linux_banners_ = linux_symbol_cache.LinuxSymbolCache.load_linux_banners() + self._linux_banners_ = symbol_cache.LinuxSymbolCache.load_banners() return self._linux_banners_ def __call__(self, @@ -124,7 +124,7 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface): if isinstance(layer, intel.Intel): return None - linux_banners = linux_symbol_cache.LinuxSymbolCache.load_linux_banners() + linux_banners = symbol_cache.LinuxSymbolCache.load_banners() mss = scanners.MultiStringScanner([x for x in linux_banners if x is not None]) for _, banner in layer.scan(context = context, scanner = mss, progress_callback = progress_callback): dtb = None diff --git a/volatility/framework/automagic/mac.py b/volatility/framework/automagic/mac.py index 8610f9183..cc9875068 100644 --- a/volatility/framework/automagic/mac.py +++ b/volatility/framework/automagic/mac.py @@ -1,17 +1,17 @@ -import sys, struct import logging +import struct import typing -import volatility.framework.objects.utility -from volatility.framework import layers, interfaces, constants, validity, exceptions -from volatility.framework import symbols, objects -from volatility.framework.automagic import mac_symbol_cache +from volatility.framework import interfaces, constants, validity +from volatility.framework import symbols +from volatility.framework.automagic import symbol_cache from volatility.framework.configuration import requirements from volatility.framework.layers import intel, scanners from volatility.framework.symbols import mac vollog = logging.getLogger(__name__) + class MacSymbolFinder(interfaces.automagic.AutomagicInterface): """Mac symbol loader based on uname signature strings""" priority = 40 @@ -24,11 +24,11 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface): self._mac_banners_ = {} # type: mac_symbol_cache.MacBanners @property - def _mac_banners(self) -> mac_symbol_cache.MacBanners: + def _mac_banners(self) -> symbol_cache.BannersType: """Creates a cached copy of the results, but only it's been requested""" if not self._mac_banners_: - self._mac_banners_ = mac_symbol_cache.MacSymbolCache.load_mac_banners() - + self._mac_banners_ = symbol_cache.MacSymbolCache.load_mac_banners() + return self._mac_banners_ def __call__(self, @@ -37,7 +37,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface): requirement: interfaces.configuration.RequirementInterface, progress_callback: validity.ProgressCallback = None) -> None: """Searches for MacSymbolRequirements and attempt to populate them""" - + self._requirements = self.find_requirements(context, config_path, requirement, (requirements.TranslationLayerRequirement, requirements.SymbolRequirement), @@ -66,7 +66,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface): progress_callback: validity.ProgressCallback = None) -> None: """Accepts a context, config_path and SymbolRequirement, with a constructed layer_name and scans the layer for mac banners""" - + # Bomb out early if there's no banners if not self._mac_banners: return @@ -83,7 +83,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface): # TODO: Fix this so it works for layers other than just Intel layer = context.memory[layer.config['memory_layer']] banner_list = layer.scan(context = context, scanner = mss, progress_callback = progress_callback) - + for _, banner in banner_list: vollog.debug("Identified banner: {}".format(repr(banner))) symbol_files = self._mac_banners.get(banner, None) @@ -126,10 +126,11 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface): if isinstance(layer, intel.Intel): return None - mac_banners = mac_symbol_cache.MacSymbolCache.load_mac_banners() + mac_banners = symbol_cache.MacSymbolCache.load_mac_banners() mss = scanners.MultiStringScanner([x for x in mac_banners if x is not None]) - - for banner_offset, banner in layer.scan(context = context, scanner = mss, progress_callback = progress_callback): + + for banner_offset, banner in layer.scan(context = context, scanner = mss, + progress_callback = progress_callback): dtb = None vollog.debug("Identified banner: {}".format(repr(banner))) @@ -138,15 +139,15 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface): isf_path = symbol_files[0] table_name = context.symbol_space.free_table_name('MacintelStacker') table = mac.MacKernelIntermedSymbols(context, 'temporary.' + table_name, name = table_name, - isf_url = isf_path) + isf_url = isf_path) context.symbol_space.append(table) kaslr_shift = MacUtilities.find_aslr(context, table_name, layer_name, - banner, banner_offset, progress_callback = progress_callback) + banner, banner_offset, progress_callback = progress_callback) ###################### # ikelos: The following is what I tried to get the dtb, but couldn't figure out how to do # as you will see after the commented block of code is just a hardcoding of the DTB to my test sample's value - ####################### + ####################### ''' bootpml4_addr = table.get_symbol("BootPML4").address + kaslr_shift @@ -168,7 +169,7 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface): print("new dtb / idlepml4_addr = {:x".format(idlepml4_addr)) sys.exit(1) ''' - + dtb = 0x1ef6e000 # Build the new layer @@ -206,10 +207,10 @@ class MacUtilities(object): def _scan_generator(self, context, layer_name, progress_callback): darwin_signature = b"Darwin Kernel Version \d{1,3}\.\d{1,3}\.\d{1,3}: [^\x00]+\x00" - + for offset in context.memory[layer_name].scan(scanner = scanners.RegExScanner(darwin_signature), context = context, progress_callback = progress_callback): - + banner = context.memory[layer_name].read(offset, 128) idx = banner.find(b"\x00") @@ -228,17 +229,17 @@ class MacUtilities(object): progress_callback: validity.ProgressCallback = None) \ -> typing.Tuple[int, int]: """Determines the offset of the actual DTB in physical space and its symbol offset""" - version_symbol = symbol_table + constants.BANG + 'version' + version_symbol = symbol_table + constants.BANG + 'version' version_json_address = context.symbol_space.get_symbol(version_symbol).address - version_phys_offset = MacUtilities.virtual_to_physical_address(version_json_address) - - version_major_symbol = symbol_table + constants.BANG + 'version_major' - version_major_json_address = context.symbol_space.get_symbol(version_major_symbol).address - version_major_phys_offset = MacUtilities.virtual_to_physical_address(version_major_json_address) + version_phys_offset = MacUtilities.virtual_to_physical_address(version_json_address) - version_minor_symbol = symbol_table + constants.BANG + 'version_minor' + version_major_symbol = symbol_table + constants.BANG + 'version_major' + version_major_json_address = context.symbol_space.get_symbol(version_major_symbol).address + version_major_phys_offset = MacUtilities.virtual_to_physical_address(version_major_json_address) + + version_minor_symbol = symbol_table + constants.BANG + 'version_minor' version_minor_json_address = context.symbol_space.get_symbol(version_minor_symbol).address - version_minor_phys_offset = MacUtilities.virtual_to_physical_address(version_minor_json_address) + version_minor_phys_offset = MacUtilities.virtual_to_physical_address(version_minor_json_address) module = context.module(symbol_table, layer_name, 0) @@ -264,20 +265,19 @@ class MacUtilities(object): minor = struct.unpack(" int: """Converts a virtual mac address to a physical one (does not account of ASLR)""" return addr - 0xffffff8000000000 - diff --git a/volatility/framework/automagic/mac_symbol_cache.py b/volatility/framework/automagic/mac_symbol_cache.py deleted file mode 100644 index 27a488dc1..000000000 --- a/volatility/framework/automagic/mac_symbol_cache.py +++ /dev/null @@ -1,99 +0,0 @@ -import logging -import os -import pickle -import typing -import urllib -import urllib.parse -import urllib.request - -from volatility.framework import constants, exceptions, interfaces -from volatility.framework.symbols import intermed - -vollog = logging.getLogger(__name__) - -MacBanners = typing.Dict[bytes, typing.List[str]] - -class MacSymbolCache(interfaces.automagic.AutomagicInterface): - """Runs through all Mac symbols tables and caches their banners""" - - # Since this is necessary for ConstructionMagic, we set a lower priority - # The user would run it eventually either way, but running it first means it can be used that run - priority = 0 - - @classmethod - def load_mac_banners(cls) -> MacBanners: - mac_banners = {} # type: MacBanners - if os.path.exists(constants.MAC_BANNERS_PATH): - with open(constants.MAC_BANNERS_PATH, "rb") as f: - # We use pickle over JSON because we're dealing with bytes objects - mac_banners.update(pickle.load(f)) - - # Remove possibilities that can't exist locally. - remove_banners = [] - for banner in mac_banners: - for path in mac_banners[banner]: - url = urllib.parse.urlparse(path) - if url.scheme == 'file' and not os.path.exists(urllib.request.url2pathname(url.path)): - vollog.log(constants.LOGLEVEL_V, - "Removing cached path {} for banner {}: file does not exist".format(path, banner)) - mac_banners[banner].remove(path) - # This is probably excessive, but it's here if we need it - # if url.scheme == 'jar': - # zip_file, zip_path = url.path.split("!") - # zip_file = urllib.parse.urlparse(zip_file).path - # if ((not os.path.exists(zip_file)) or (zip_path not in zipfile.ZipFile(zip_file).namelist())): - # vollog.log(constants.LOGLEVEL_V, - # "Removing cached path {} for banner {}: file does not exist".format(path, banner)) - # mac_banners[banner].remove(path) - - if not mac_banners[banner]: - remove_banners.append(banner) - for remove_banner in remove_banners: - del mac_banners[remove_banner] - return mac_banners - - @classmethod - def save_mac_banners(cls, mac_banners): - - with open(constants.MAC_BANNERS_PATH, "wb") as f: - pickle.dump(mac_banners, f) - - def __call__(self, context, config_path, configurable, progress_callback = None): - """Runs the automagic over the configurable""" - # We only need to be called once, so no recursion necessary - macbanners = self.load_mac_banners() - - cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url("mac")) - vollog.info("Building mac cacheables...".format(cacheables)) - - for banner in macbanners: - for json_file in macbanners[banner]: - if json_file in cacheables: - cacheables.remove(json_file) - - total = len(cacheables) - if total > 0: - vollog.info("Building mac caches...") - for current in range(total): - #progress_callback(current * 100 / total, "Building mac caches") - isf_url = cacheables[current] - - try: - # Loading the symbol table will be very slow until it's been validated - isf = intermed.IntermediateSymbolTable(context, config_path, "temp", isf_url, validate = False) - - # We should store the banner against the filename - # We don't bother with the hash (it'll likely take too long to validate) - # but we should check at least that the banner matches on load. - banner = isf.get_symbol("version").constant_data - vollog.log(constants.LOGLEVEL_V, "Caching banner {} for file {}".format(banner, isf_url)) - bannerlist = macbanners.get(banner, []) - bannerlist.append(isf_url) - macbanners[banner] = bannerlist - except exceptions.SymbolError: - pass - - vollog.debug("writing mac banners: {}".format(macbanners)) - - # Rewrite the cached macbanners each run, since writing is faster than the cache validation portion - self.save_mac_banners(macbanners) diff --git a/volatility/framework/automagic/linux_symbol_cache.py b/volatility/framework/automagic/symbol_cache.py similarity index 60% rename from volatility/framework/automagic/linux_symbol_cache.py rename to volatility/framework/automagic/symbol_cache.py index cbe8ca49d..a170c5a5c 100644 --- a/volatility/framework/automagic/linux_symbol_cache.py +++ b/volatility/framework/automagic/symbol_cache.py @@ -11,33 +11,36 @@ from volatility.framework.symbols import intermed vollog = logging.getLogger(__name__) -LinuxBanners = typing.Dict[bytes, typing.List[str]] +BannersType = typing.Dict[bytes, typing.List[str]] -class LinuxSymbolCache(interfaces.automagic.AutomagicInterface): - """Runs through all Linux symbols tables and caches their banners""" +class SymbolCache(interfaces.automagic.AutomagicInterface): + """Runs through all symbols tables and caches their banners""" # Since this is necessary for ConstructionMagic, we set a lower priority # The user would run it eventually either way, but running it first means it can be used that run priority = 0 + os = None + symbol_name = "linux_banner" + banner_path = constants.LINUX_BANNERS_PATH @classmethod - def load_linux_banners(cls) -> LinuxBanners: - linux_banners = {} # type: LinuxBanners - if os.path.exists(constants.LINUX_BANNERS_PATH): - with open(constants.LINUX_BANNERS_PATH, "rb") as f: + def load_banners(cls) -> BannersType: + banners = {} # type: BannersType + if os.path.exists(cls.banner_path): + with open(cls.banner_path, "rb") as f: # We use pickle over JSON because we're dealing with bytes objects - linux_banners.update(pickle.load(f)) + banners.update(pickle.load(f)) # Remove possibilities that can't exist locally. remove_banners = [] - for banner in linux_banners: - for path in linux_banners[banner]: + for banner in banners: + for path in banners[banner]: url = urllib.parse.urlparse(path) if url.scheme == 'file' and not os.path.exists(urllib.request.url2pathname(url.path)): vollog.log(constants.LOGLEVEL_V, "Removing cached path {} for banner {}: file does not exist".format(path, banner)) - linux_banners[banner].remove(path) + banners[banner].remove(path) # This is probably excessive, but it's here if we need it # if url.scheme == 'jar': # zip_file, zip_path = url.path.split("!") @@ -45,37 +48,40 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface): # if ((not os.path.exists(zip_file)) or (zip_path not in zipfile.ZipFile(zip_file).namelist())): # vollog.log(constants.LOGLEVEL_V, # "Removing cached path {} for banner {}: file does not exist".format(path, banner)) - # linux_banners[banner].remove(path) + # banners[banner].remove(path) - if not linux_banners[banner]: + if not banners[banner]: remove_banners.append(banner) for remove_banner in remove_banners: - del linux_banners[remove_banner] - return linux_banners + del banners[remove_banner] + return banners @classmethod - def save_linux_banners(cls, linux_banners): + def save_banners(cls, banners): - with open(constants.LINUX_BANNERS_PATH, "wb") as f: - pickle.dump(linux_banners, f) + with open(cls.banner_path, "wb") as f: + pickle.dump(banners, f) def __call__(self, context, config_path, configurable, progress_callback = None): """Runs the automagic over the configurable""" + if self.os is None: + return + # We only need to be called once, so no recursion necessary - linuxbanners = self.load_linux_banners() + banners = self.load_banners() - cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url("linux")) + cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url(self.os)) - for banner in linuxbanners: - for json_file in linuxbanners[banner]: + for banner in banners: + for json_file in banners[banner]: if json_file in cacheables: cacheables.remove(json_file) total = len(cacheables) if total > 0: - vollog.info("Building linux caches...") + vollog.info("Building {} caches...".format(self.os)) for current in range(total): - progress_callback(current * 100 / total, "Building linux caches") + progress_callback(current * 100 / total, "Building {} caches".format(self.os)) isf_url = cacheables[current] try: @@ -85,13 +91,25 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface): # We should store the banner against the filename # We don't bother with the hash (it'll likely take too long to validate) # but we should check at least that the banner matches on load. - banner = isf.get_symbol("linux_banner").constant_data + banner = isf.get_symbol(self.symbol_name).constant_data vollog.log(constants.LOGLEVEL_V, "Caching banner {} for file {}".format(banner, isf_url)) - bannerlist = linuxbanners.get(banner, []) + bannerlist = banners.get(banner, []) bannerlist.append(isf_url) - linuxbanners[banner] = bannerlist + banners[banner] = bannerlist except exceptions.SymbolError: pass - # Rewrite the cached linuxbanners each run, since writing is faster than the cache validation portion - self.save_linux_banners(linuxbanners) + # Rewrite the cached banners each run, since writing is faster than the cache validation portion + self.save_banners(banners) + + +class LinuxSymbolCache(SymbolCache): + os = "linux" + symbol_name = "linux_banner" + banner_path = constants.LINUX_BANNERS_PATH + + +class MacSymbolCache(SymbolCache): + os = "mac" + symbol_name = "version" + banner_path = constants.MAC_BANNERS_PATH