From 79b3be4cf6a2f11602741e7e83269a9bdae07c11 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sun, 13 Jun 2021 21:32:03 +0100 Subject: [PATCH] Automagic: Add in remote banner cache first attempt --- development/banner_server.py | 77 +++++++++++++++++++ .../framework/automagic/symbol_cache.py | 71 ++++++++++++++--- volatility3/framework/constants/__init__.py | 6 ++ 3 files changed, 142 insertions(+), 12 deletions(-) create mode 100644 development/banner_server.py diff --git a/development/banner_server.py b/development/banner_server.py new file mode 100644 index 000000000..3449babdd --- /dev/null +++ b/development/banner_server.py @@ -0,0 +1,77 @@ +import argparse +import base64 +import json +import logging +import os +import pathlib +import urllib + +from volatility3.cli import PrintedProgress +from volatility3.framework import contexts, constants +from volatility3.framework.automagic import linux, mac + +vollog = logging.getLogger(__name__) + + +class BannerCacheGenerator: + + def __init__(self, path: str, url_prefix: str): + self._path = path + self._url_prefix = url_prefix + + def convert_url(self, url): + parsed = urllib.parse.urlparse(url) + + relpath = os.path.relpath(parsed.path, os.path.abspath(self._path)) + + return urllib.parse.urljoin(self._url_prefix, relpath) + + def run(self): + context = contexts.Context() + json_output = {} + + path = self._path + filename = '*' + + for banner_cache in [linux.LinuxBannerCache, mac.MacBannerCache]: + sub_path = banner_cache.os + potentials = [] + for extension in constants.ISF_EXTENSIONS: + # Hopefully these will not be large lists, otherwise this might be slow + try: + for found in pathlib.Path(path).joinpath(sub_path).resolve().rglob(filename + extension): + potentials.append(found.as_uri()) + except FileNotFoundError: + # If there's no linux symbols, don't cry about it + pass + + new_banners = banner_cache.read_new_banners(context, 'BannerServer', potentials, banner_cache.symbol_name, + banner_cache.os, progress_callback = PrintedProgress()) + result_banners = {} + for new_banner in new_banners: + # Only accept file schemes + value = [self.convert_url(url) for url in new_banners[new_banner] if + urllib.parse.urlparse(url).scheme == 'file'] + if value and new_banner: + # Convert files into URLs + result_banners[str(base64.b64encode(new_banner), 'latin-1')] = value + + json_output[banner_cache.os] = result_banners + + output_path = os.path.join(self._path, 'banners.json') + with open(output_path, 'w') as fp: + vollog.warning(f"Banners file written to {output_path}") + json.dump(json_output, fp) + + +if __name__ == '__main__': + + parser = argparse.ArgumentParser() + parser.add_argument('--path', default = os.path.dirname(__file__)) + parser.add_argument('--urlprefix', help = 'Web prefix that will eventually serve the ISF files', + default = 'http://localhost/symbols') + + args = parser.parse_args() + + bcg = BannerCacheGenerator(args.path, args.urlprefix) + bcg.run() diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index 627ffac8b..59999a07e 100644 --- a/volatility3/framework/automagic/symbol_cache.py +++ b/volatility3/framework/automagic/symbol_cache.py @@ -1,6 +1,7 @@ # 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 base64 import gc import json import logging @@ -9,9 +10,11 @@ import pickle import urllib import urllib.parse import urllib.request +import zipfile from typing import Dict, List, Optional from volatility3.framework import constants, exceptions, interfaces +from volatility3.framework.layers import resources from volatility3.framework.symbols import intermed vollog = logging.getLogger(__name__) @@ -81,20 +84,42 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface): # We only need to be called once, so no recursion necessary banners = self.load_banners() - cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url(self.os)) + cacheables = self.find_new_banner_files(banners, self.os) - for banner in banners: - for json_file in banners[banner]: - if json_file in cacheables: - cacheables.remove(json_file) + new_banners = self.read_new_banners(context, config_path, cacheables, self.symbol_name, self.os, + progress_callback) - total = len(cacheables) + # Add in any new banners to the existing list + for new_banner in new_banners: + banner_list = banners.get(new_banner, []) + banners[new_banner] = list(set(banner_list + new_banners[new_banner])) + + # Do remote banners *after* the JSON loading, so that it doen't pull down all the remote JSON + self.remote_banners(banners, self.os) + + # Rewrite the cached banners each run, since writing is faster than the banner_cache validation portion + self.save_banners(banners) + + if progress_callback is not None: + progress_callback(100, "Built {} caches".format(self.os)) + + @classmethod + def read_new_banners(cls, context: interfaces.context.ContextInterface, config_path: str, new_urls: List[str], + symbol_name: str, operating_system: str = None, + progress_callback = None) -> Optional[Dict[bytes, List[str]]]: + """Reads the any new banners for the OS in question""" + if operating_system is None: + return None + + banners = {} + + total = len(new_urls) if total > 0: vollog.info(f"Building {self.os} caches...") for current in range(total): if progress_callback is not None: progress_callback(current * 100 / total, f"Building {self.os} caches") - isf_url = cacheables[current] + isf_url = new_urls[current] isf = None try: @@ -104,7 +129,7 @@ class SymbolBannerCache(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(self.symbol_name).constant_data + banner = isf.get_symbol(symbol_name).constant_data vollog.log(constants.LOGLEVEL_VV, f"Caching banner {banner} for file {isf_url}") bannerlist = banners.get(banner, []) @@ -119,9 +144,31 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface): if isf: del isf gc.collect() + return banners - # Rewrite the cached banners each run, since writing is faster than the banner_cache validation portion - self.save_banners(banners) + @classmethod + def find_new_banner_files(cls, banners: Dict[bytes, List[str]], operating_system: str) -> List[str]: + """Gathers all files and remove existing banners""" + cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url(operating_system)) + for banner in banners: + for json_file in banners[banner]: + if json_file in cacheables: + cacheables.remove(json_file) + return cacheables - if progress_callback is not None: - progress_callback(100, f"Built {self.os} caches") + @classmethod + def remote_banners(cls, banners: Dict[bytes, List[str]], operating_system = None): + """Adds remote URLs to the banner list""" + if operating_system is None: + return None + + if not constants.OFFLINE: + # TODO: Only download the remote file once per amount of time + with resources.ResourceAccessor().open(url = constants.REMOTE_ISF_URL) as fp: + banner_list = json.load(fp) + if operating_system in banner_list: + for banner in banner_list[operating_system]: + binary_banner = base64.b64decode(banner) + file_list = banners.get(binary_banner, []) + file_list = list(set(file_list + banner_list[operating_system][banner])) + banners[binary_banner] = file_list diff --git a/volatility3/framework/constants/__init__.py b/volatility3/framework/constants/__init__.py index b85c52d16..9ec2fb88f 100644 --- a/volatility3/framework/constants/__init__.py +++ b/volatility3/framework/constants/__init__.py @@ -94,3 +94,9 @@ ISF_MINIMUM_SUPPORTED = (2, 0, 0) """The minimum supported version of the Intermediate Symbol Format""" ISF_MINIMUM_DEPRECATED = (3, 9, 9) """The highest version of the ISF that's deprecated (usually higher than supported)""" + +OFFLINE = False +"""Whether to go online to retrieve missing/necessary JSON files""" + +REMOTE_ISF_URL = 'http://localhost:8000/banners.json' +"""Remote URL to query for a list of ISF addresses"""