diff --git a/development/banner_server.py b/development/banner_server.py new file mode 100644 index 000000000..3aea41c82 --- /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 = {'version': 1} + + 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/cli/__init__.py b/volatility3/cli/__init__.py index e6f00728a..6d75db46c 100644 --- a/volatility3/cli/__init__.py +++ b/volatility3/cli/__init__.py @@ -165,6 +165,10 @@ class CommandLine: help = f"Change the default path ({constants.CACHE_PATH}) used to store the cache", default = constants.CACHE_PATH, type = str) + parser.add_argument("--offline", + help = "Do not search online for additional JSON files", + default = False, + action = 'store_true') # We have to filter out help, otherwise parse_known_args will trigger the help message before having # processed the plugin choice or had the plugin subparser added. @@ -216,6 +220,9 @@ class CommandLine: if partial_args.clear_cache: framework.clear_cache() + if partial_args.offline: + constants.OFFLINE = partial_args.offline + # Do the initialization ctx = contexts.Context() # Construct a blank context failures = framework.import_files(volatility3.plugins, diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index 627ffac8b..b2c227228 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__) @@ -51,13 +54,13 @@ class SymbolBannerCache(interfaces.automagic.AutomagicInterface): path, str(banner or b'', 'latin-1'))) 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_VV, - # "Removing cached path {} for banner {}: file does not exist".format(path, banner)) - # banners[banner].remove(path) + 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_VV, + "Removing cached path {} for banner {}: file does not exist".format(path, banner)) + banners[banner].remove(path) if not banners[banner]: remove_banners.append(banner) @@ -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, f"Built {self.os} caches") + + @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...") + vollog.info(f"Building {operating_system} 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] + progress_callback(current * 100 / total, f"Building {operating_system} caches") + 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,67 @@ 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, banner_location = None): + """Adds remote URLs to the banner list""" + if operating_system is None: + return None + + if banner_location is None: + banner_location = constants.REMOTE_ISF_URL + + if not constants.OFFLINE and banner_location is not None: + try: + rbf = RemoteBannerFormat(banner_location) + rbf.process(banners, operating_system) + except urllib.error.URLError: + vollog.debug(f"Unable to download remote banner list from {banner_location}") + + +class RemoteBannerFormat: + def __init__(self, location: str): + self._location = location + with resources.ResourceAccessor().open(url = location) as fp: + self._data = json.load(fp) + if not self._verify(): + raise ValueError("Unsupported version for remote banner list format") + + def _verify(self) -> bool: + version = self._data.get('version', 0) + if version in [1]: + setattr(self, 'process', getattr(self, f'process_v{version}')) + return True + return False + + def process(self, banners: Dict[bytes, List[str]], operating_system: Optional[str]): + raise ValueError("Banner List version not verified") + + def process_v1(self, banners: Dict[bytes, List[str]], operating_system: Optional[str]): + if operating_system in self._data: + for banner in self._data[operating_system]: + binary_banner = base64.b64decode(banner) + file_list = banners.get(binary_banner, []) + for value in self._data[operating_system][banner]: + if value not in file_list: + file_list = file_list + [value] + banners[binary_banner] = file_list + if 'additional' in self._data: + for location in self._data['additional']: + try: + subrbf = RemoteBannerFormat(location) + subrbf.process(banners, operating_system) + except IOError: + vollog.debug(f"Remote file not found: {location}") + return banners diff --git a/volatility3/framework/constants/__init__.py b/volatility3/framework/constants/__init__.py index cfe0356c1..43193852a 100644 --- a/volatility3/framework/constants/__init__.py +++ b/volatility3/framework/constants/__init__.py @@ -40,7 +40,7 @@ BANG = "!" # We use the SemVer 2.0.0 versioning scheme VERSION_MAJOR = 1 # Number of releases of the library with a breaking change VERSION_MINOR = 2 # Number of changes that only add to the interface -VERSION_PATCH = 0 # Number of changes that do not change the interface +VERSION_PATCH = 1 # Number of changes that do not change the interface VERSION_SUFFIX = "" # TODO: At version 2.0.0, remove the symbol_shift feature @@ -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 = None # 'http://localhost:8000/banners.json' +"""Remote URL to query for a list of ISF addresses""" diff --git a/volatility3/framework/exceptions.py b/volatility3/framework/exceptions.py index 7f381b166..a234a353a 100644 --- a/volatility3/framework/exceptions.py +++ b/volatility3/framework/exceptions.py @@ -99,3 +99,14 @@ class MissingModuleException(VolatilityException): def __init__(self, module: str, *args) -> None: super().__init__(*args) self.module = module + + +class OfflineException(VolatilityException): + """Throw when a remote resource is requested but Volatility is in offline mode""" + + def __init__(self, url: str, *args) -> None: + super().__init__(*args) + self._url = url + + def __str__(self): + return f'Volatility 3 is offline: unable to access {self._url}' diff --git a/volatility3/framework/layers/resources.py b/volatility3/framework/layers/resources.py index fb8bdb7cc..35182a86b 100644 --- a/volatility3/framework/layers/resources.py +++ b/volatility3/framework/layers/resources.py @@ -17,7 +17,7 @@ from typing import Optional, Any, IO, List from urllib import error from volatility3 import framework -from volatility3.framework import constants +from volatility3.framework import constants, exceptions try: import magic @@ -34,6 +34,7 @@ except ImportError: vollog = logging.getLogger(__name__) + # TODO: Type-annotating the ResourceAccessor.open method is difficult because HTTPResponse is not actually an IO[Any] type # fix this @@ -117,6 +118,9 @@ class ResourceAccessor(object): raise excp else: raise excp + except exceptions.OfflineException: + vollog.info(f"Not accessing {url} in offline mode") + raise with contextlib.closing(fp) as fp: # Cache the file locally @@ -227,6 +231,7 @@ class JarHandler(VolatilityHandler): Actual reference (found from https://www.w3.org/wiki/UriSchemes/jar) seemed not to return: http://developer.java.sun.com/developer/onlineTraining/protocolhandlers/ """ + @classmethod def non_cached_schemes(cls) -> List[str]: return ['jar'] @@ -249,3 +254,11 @@ class JarHandler(VolatilityHandler): zippath, filepath = zipsplit return zipfile.ZipFile(zippath).open(filepath) return None + + +class OfflineHandler(VolatilityHandler): + @staticmethod + def default_open(req: urllib.request.Request) -> Optional[Any]: + if constants.OFFLINE and req.type in ['http', 'https']: + raise exceptions.OfflineException(req.full_url) + return None