Merge pull request #526 from volatilityfoundation/feature/linux-mac-repo

Feature/linux mac repo

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()

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,

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

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"""

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}'

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