Automagic: Use sqlite to cache identifiers

volatility3/framework/automagic/linux.py

@@ -5,8 +5,9 @@
 import logging
 from typing import Optional, Tuple, Type
 
-from volatility3.framework import interfaces, constants
+from volatility3.framework import constants, interfaces
 from volatility3.framework.automagic import symbol_cache, symbol_finder
+from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import intel, scanners
 from volatility3.framework.symbols import linux
 
@@ -23,6 +24,13 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
               layer_name: str,
               progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]:
         """Attempts to identify linux within this layer."""
+        # Version check the SQlite cache
+        required = (1, 0, 0)
+        if not requirements.VersionRequirement.matches_required(required, symbol_cache.SqliteCache.version):
+            vollog.info(
+                f"SQLiteCache version not suitable: required {required} found {symbol_cache.SqliteCache.version}")
+            return None
+
         # Bail out by default unless we can stack properly
         layer = context.layers[layer_name]
         join = interfaces.configuration.path_join
@@ -32,7 +40,8 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
         if isinstance(layer, intel.Intel):
             return None
 
-        linux_banners = LinuxBannerCache.load_banners()
+        linux_banners = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH).get_identifier_dictionary(
+            operating_system = 'linux')
         # If we have no banners, don't bother scanning
         if not linux_banners:
             vollog.info("No Linux banners found - if this is a linux plugin, please check your symbol files location")
@@ -43,15 +52,8 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
             dtb = None
             vollog.debug(f"Identified banner: {repr(banner)}")
 
-            symbol_files = linux_banners.get(banner, None)
-            if symbol_files:
-                if len(symbol_files) > 1:
-                    using = "*"
-                    vollog.warning(f"Multiple symbol files identified (using {using}):")
-                    for symbol_file in symbol_files:
-                        vollog.warning(f" {using} {symbol_file}")
-                        using = " "
-                isf_path = symbol_files[0]
+            isf_path = linux_banners.get(banner, None)
+            if isf_path:
                 table_name = context.symbol_space.free_table_name('LintelStacker')
                 table = linux.LinuxKernelIntermedSymbols(context,
                                                          'temporary.' + table_name,
@@ -147,20 +149,11 @@ class LinuxIntelStacker(interfaces.automagic.StackerLayerInterface):
         return addr - 0xc0000000
 
 
-class LinuxBannerCache(symbol_cache.SymbolBannerCache):
-    """Caches the banners found in the Linux symbol files."""
-
-    os = "linux"
-    symbol_name = "linux_banner"
-    banner_path = constants.LINUX_BANNERS_PATH
-    exclusion_list = ['mac', 'windows']
-
-
 class LinuxSymbolFinder(symbol_finder.SymbolFinder):
     """Linux symbol loader based on uname signature strings."""
 
     banner_config_key = "kernel_banner"
-    banner_cache = LinuxBannerCache
+    operating_system = 'linux'
     symbol_class = "volatility3.framework.symbols.linux.LinuxKernelIntermedSymbols"
     find_aslr = lambda cls, *args: LinuxIntelStacker.find_aslr(*args)[1]
     exclusion_list = ['mac', 'windows']

volatility3/framework/automagic/mac.py

@@ -6,8 +6,9 @@ import logging
 import struct
 from typing import Optional
 
-from volatility3.framework import interfaces, constants, layers, exceptions
+from volatility3.framework import constants, exceptions, interfaces, layers
 from volatility3.framework.automagic import symbol_cache, symbol_finder
+from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import intel, scanners
 from volatility3.framework.symbols import mac
 
@@ -24,6 +25,13 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
               layer_name: str,
               progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]:
         """Attempts to identify mac within this layer."""
+        # Version check the SQlite cache
+        required = (1, 0, 0)
+        if not requirements.VersionRequirement.matches_required(required, symbol_cache.SqliteCache.version):
+            vollog.info(
+                f"SQLiteCache version not suitable: required {required} found {symbol_cache.SqliteCache.version}")
+            return None
+
         # Bail out by default unless we can stack properly
         layer = context.layers[layer_name]
         new_layer = None
@@ -34,7 +42,8 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
         if isinstance(layer, intel.Intel):
             return None
 
-        mac_banners = MacBannerCache.load_banners()
+        mac_banners = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH).get_identifier_dictionary(
+            operating_system = 'mac')
         # If we have no banners, don't bother scanning
         if not mac_banners:
             vollog.info("No Mac banners found - if this is a mac plugin, please check your symbol files location")
@@ -46,9 +55,8 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
             dtb = None
             vollog.debug(f"Identified banner: {repr(banner)}")
 
-            symbol_files = mac_banners.get(banner, None)
-            if symbol_files:
-                isf_path = symbol_files[0]
+            isf_path = mac_banners.get(banner, None)
+            if isf_path:
                 table_name = context.symbol_space.free_table_name('MacintelStacker')
                 table = mac.MacKernelIntermedSymbols(context = context,
                                                      config_path = join('temporary', table_name),
@@ -197,19 +205,11 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
             yield offset, banner
 
 
-class MacBannerCache(symbol_cache.SymbolBannerCache):
-    """Caches the banners found in the Mac symbol files."""
-    os = "mac"
-    symbol_name = "version"
-    banner_path = constants.MAC_BANNERS_PATH
-    exclusion_list = ['windows', 'linux']
-
-
 class MacSymbolFinder(symbol_finder.SymbolFinder):
     """Mac symbol loader based on uname signature strings."""
 
     banner_config_key = 'kernel_banner'
-    banner_cache = MacBannerCache
+    operating_system = 'mac'
     find_aslr = MacIntelStacker.find_aslr
     symbol_class = "volatility3.framework.symbols.mac.MacKernelIntermedSymbols"
     exclusion_list = ['windows', 'linux']

volatility3/framework/automagic/symbol_cache.py

@@ -2,18 +2,20 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 import base64
-import gc
 import json
 import logging
 import os
-import pickle
+import sqlite3
 import urllib
 import urllib.parse
 import urllib.request
-import zipfile
-from typing import Dict, List, Optional
+from abc import abstractmethod
+from typing import Dict, Generator, List, Optional
 
-from volatility3.framework import constants, exceptions, interfaces
+import volatility3.framework
+import volatility3.schemas
+from volatility3.framework import constants, interfaces
+from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import resources
 from volatility3.framework.symbols import intermed
 
@@ -22,164 +24,324 @@ vollog = logging.getLogger(__name__)
 BannersType = Dict[bytes, List[str]]
 
 
-class SymbolBannerCache(interfaces.automagic.AutomagicInterface):
-    """Runs through all symbols tables and caches their banners."""
+### Identifiers
 
-    # 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
+class IdentifierProcessor:
+    operating_system = None
+
+    def __init__(self):
+        pass
+
+    @classmethod
+    @abstractmethod
+    def get_identifier(cls, json) -> Optional[bytes]:
+        """Method to extract the identifier from a particular operating system's JSON
+
+        Returns:
+            identifier is valid or None if not found
+        """
+        raise NotImplemented("This base class has no get_identifier method defined")
+
+
+class WindowsIdentifier(IdentifierProcessor):
+    operating_system = 'windows'
+    separator = '|'
+
+    @classmethod
+    def get_identifier(cls, json) -> Optional[bytes]:
+        """Returns the identifier for the file if one can be found"""
+        windows_metadata = json.get('metadata', {}).get('windows', {}).get('pdb', {})
+        if windows_metadata:
+            guid = windows_metadata.get('GUID', None)
+            age = windows_metadata.get('age', None)
+            database = windows_metadata.get('database', None)
+            if guid and age and database:
+                return cls.generate(database, guid, age)
+        return None
+
+    @classmethod
+    def generate(cls, pdb_name: str, guid: str, age: int) -> bytes:
+        return bytes(cls.separator.join([pdb_name, guid.upper(), str(age)]), 'latin-1')
+
+
+class MacIdentifier(IdentifierProcessor):
+    operating_system = 'mac'
+
+    @classmethod
+    def get_identifier(cls, json) -> Optional[bytes]:
+        mac_banner = json.get('symbols', {}).get('version', {}).get('constant_data', None)
+        if mac_banner:
+            return base64.b64decode(mac_banner)
+        return None
+
+
+class LinuxIdentifier(IdentifierProcessor):
+    operating_system = 'linux'
+
+    @classmethod
+    def get_identifier(cls, json) -> Optional[bytes]:
+        linux_banner = json.get('symbols', {}).get('linux_banner', {}).get('constant_data', None)
+        if linux_banner:
+            return base64.b64decode(linux_banner)
+        return None
+
+
+### CacheManagers
+
+class CacheManagerInterface(interfaces.configuration.VersionableInterface):
+    def __init__(self, filename: str):
+        super().__init__()
+        self._filename = filename
+        self._classifiers = {}
+        for subclazz in volatility3.framework.class_subclasses(IdentifierProcessor):
+            self._classifiers[subclazz.operating_system] = subclazz
+
+    def add_identifier(self, location: str, operating_system: str, identifier: str):
+        """Adds an identifier to the store"""
+        pass
+
+    def find_location(self, identifier: bytes, operating_system: Optional[str]) -> Optional[str]:
+        """Returns the location of the symbol file given the identifier
+
+        Args:
+            identifier: string that uniquely identifies a particular symbolt table
+            operating_system: optional string to restrict identifiers to just those for a particular operating system
+
+        Returns:
+            The location of the symbols file that matches the identifier
+        """
+        pass
+
+    def get_local_locations(self) -> List[str]:
+        """Returns a list of all the local locations"""
+        pass
+
+    def update(self):
+        """Locates all files under the symbol directories.  Updates the cache with additions, modifications and removals.
+        This also updates remote locations based on a cache timeout.
+
+        """
+        pass
+
+    def get_identifier_dictionary(self, operating_system: Optional[str] = None, local_only: bool = False) -> \
+            Dict[bytes, str]:
+        """Returns a dictionary of identifiers and locations
+
+        Args:
+            operating_system: If set, limits responses to a specific operating system
+            local_only: Returns only local locations
+
+        Returns:
+            A dictionary of identifiers mapped to a location
+        """
+        pass
+
+    def get_identifier(self, location: str) -> Optional[bytes]:
+        """Returns an identifier based on a specific location or None"""
+        pass
+
+    def get_identifiers(self, operating_system: Optional[str]):
+        """Returns all identifiers for a particular operating system"""
+        pass
+
+
+class SqliteCache(CacheManagerInterface):
+    _required_framework_version = (2, 0, 0)
+    _version = (1, 0, 0)
+
+    def __init__(self, filename: str):
+        super().__init__(filename)
+        try:
+            self._database = self._connect_storage(filename)
+        except sqlite3.DatabaseError:
+            os.unlink(filename)
+            self._database = self._connect_storage(filename)
+
+    def _connect_storage(self, path: str):
+        database = sqlite3.connect(path, isolation_level = None)
+        database.row_factory = sqlite3.Row
+        database.cursor().execute(
+            'CREATE TABLE IF NOT EXISTS cache (location TEXT UNIQUE NOT NULL, identifier TEXT, operating_system TEXT, local BOOL, cached DATETIME)')
+        return database
+
+    def find_location(self, identifier: bytes, operating_system: Optional[str]) -> Optional[str]:
+        """Returns the location of the symbol file given the identifier.
+        If multiple locations exist for an identifier, the last found is returned
+
+        Args:
+            identifier: string that uniquely identifies a particular symbolt table
+            operating_system: optional string to restrict identifiers to just those for a particular operating system
+
+        Returns:
+            The location of the symbols file that matches the identifier or None
+        """
+        statement = 'SELECT location FROM cache WHERE identifier = ?'
+        parameters = (identifier,)
+        if operating_system is not None:
+            statement = 'SELECT location FROM cache WHERE identifier = ? AND operating_system = ?'
+            parameters = (identifier, operating_system)
+        results = self._database.cursor().execute(statement, parameters).fetchall()
+        result = None
+        for row in results:
+            result = row['location']
+        return result
+
+    def get_local_locations(self) -> Generator[str, None, None]:
+        result = self._database.cursor().execute('SELECT DISTINCT location FROM cache WHERE local = True').fetchall()
+        for row in result:
+            yield row['location']
+
+    def is_url_local(self, url: str) -> bool:
+        """Determines whether an url is local or not"""
+        parsed = urllib.parse.urlparse(url)
+        if parsed.scheme in ['file', 'jar']:
+            return True
+
+    def get_identifier(self, location: str) -> Optional[bytes]:
+        results = self._database.cursor().execute('SELECT identifier FROM cache WHERE location = ?',
+                                                  (location,)).fetchall()
+        for row in results:
+            return row['identifier']
+        return None
+
+    def update(self, progress_callback = None):
+        """Locates all files under the symbol directories.  Updates the cache with additions, modifications and removals.
+        This also updates remote locations based on a cache timeout.
+
+        """
+        on_disk_locations = set([filename for filename in intermed.IntermediateSymbolTable.file_symbol_url('')])
+        cached_locations = set(self.get_local_locations())
+
+        new_locations = on_disk_locations.difference(cached_locations)
+        missing_locations = cached_locations.difference(on_disk_locations)
+
+        cache_update = set()
+        files_to_timestamp = on_disk_locations.intersection(cached_locations)
+        if files_to_timestamp:
+            result = self._database.cursor().execute("SELECT location FROM cache WHERE local = True "
+                                                     "AND cached < date('now', '-3 days');")
+            for row in result:
+                if row['location'] in files_to_timestamp:
+                    cache_update.add(row['location'])
+
+        idextractors = list(volatility3.framework.class_subclasses(IdentifierProcessor))
+
+        counter = 0
+        files_to_process = new_locations.union(cache_update)
+        number_files_to_process = len(files_to_process)
+        for location in files_to_process:
+            # Open location
+            counter += 1
+            progress_callback(counter * 100 / number_files_to_process,
+                              "Updating caches for {number_files_to_process} files...")
+            try:
+                with resources.ResourceAccessor().open(location) as fp:
+                    json_obj = json.load(fp)
+                    identifier = None
+                    for idextractor in idextractors:
+                        identifier = idextractor.get_identifier(json_obj)
+                        operating_system = idextractor.operating_system
+                        if identifier is not None:
+                            break
+                    if identifier is not None:
+                        # We don't try to validate schemas here, we do that on first use
+                        # Store in database
+                        self._database.cursor().execute(
+                            "INSERT OR REPLACE INTO cache (location, identifier, operating_system, local, cached) VALUES (?, ?, ?, ?, datetime('now'))",
+                            (
+                                location,
+                                identifier,
+                                operating_system,
+                                self.is_url_local(location)
+                            ))
+                        vollog.log(constants.LOGLEVEL_VV, f"Identified {location} as {identifier}")
+                    else:
+                        self._database.cursor().execute(
+                            "INSERT OR REPLACE INTO cache (location, identifier, operating_system, local, cached) VALUES (?, ?, ?, ?, datetime('now'))",
+                            (
+                                location,
+                                None,
+                                None,
+                                self.is_url_local(location)
+                            ))
+                        vollog.log(constants.LOGLEVEL_VVVV, f"No identifier found for {location}")
+            except Exception as excp:
+                vollog.log(constants.LOGLEVEL_VVVV, excp)
+
+        if not constants.OFFLINE and constants.REMOTE_ISF_URL:
+            remote_identifiers = RemoteIdentifierFormat(constants.REMOTE_ISF_URL)
+            for operating_system in ['mac', 'linux', 'windows']:
+                identifiers = remote_identifiers.process({}, operating_system = operating_system)
+                for identifier in identifiers:
+                    for location in identifiers[identifier]:
+                        self._database.cursor().execute(
+                            "INSERT OR REPLACE INTO cache(identifier, location, operating_system, local, cached) VALUES (?, ?, ?, ?, datetime('now')",
+                            (location, identifier, operating_system, False)
+                        )
+
+        if missing_locations:
+            self._database.cursor().execute(
+                f"DELETE FROM cache WHERE location IN ({','.join(['?'] * len(missing_locations))})", *missing_locations)
+
+    def get_identifier_dictionary(self, operating_system: Optional[str] = None, local_only: bool = False) -> \
+            Dict[bytes, str]:
+        output = {}
+        additions = []
+        statement = 'SELECT location, identifier FROM cache'
+        if local_only:
+            additions.append('local = True')
+        if operating_system:
+            additions.append(f"operating_system = '{operating_system}'")
+        if additions:
+            statement += f" WHERE {' AND '.join(additions)}"
+        results = self._database.cursor().execute(statement)
+        for row in results:
+            if row['identifier'] in output and row['identifier'] and row['location']:
+                vollog.debug(
+                    f"Duplicate entry for identifier {row['identifier']}: {row['location']} and {output[row['identifier']]}")
+            output[row['identifier']] = row['location']
+        return output
+
+    def get_identifiers(self, operating_system: Optional[str]):
+        if operating_system:
+            results = self._database.cursor().execute('SELECT identifier FROM cache WHERE operating_system = ?',
+                                                      (operating_system,)).fetchall()
+        else:
+            results = self._database.cursor().execute('SELECT identifier FROM cache').fetchall()
+        output = []
+        for row in results:
+            output.append(row['identifier'])
+        return output
+
+
+### Automagic
+
+class SymbolCacheMagic(interfaces.automagic.AutomagicInterface):
+    """Runs through all symbol tables and caches their identifiers"""
     priority = 0
 
-    os: Optional[str] = None
-    symbol_name: str = "banner_name"
-    banner_path: Optional[str] = None
-
-    @classmethod
-    def load_banners(cls) -> BannersType:
-        if not cls.banner_path:
-            raise ValueError("Banner_path not appropriately set")
-        banners: 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
-                banners.update(pickle.load(f))
-
-        # Remove possibilities that can't exist locally.
-        remove_banners = []
-        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_VV, "Removing cached path {} for banner {}: file does not exist".format(
-                            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 not banners[banner]:
-                remove_banners.append(banner)
-        for remove_banner in remove_banners:
-            del banners[remove_banner]
-        return banners
-
-    @classmethod
-    def save_banners(cls, banners):
-
-        with open(cls.banner_path, "wb") as f:
-            pickle.dump(banners, f)
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._cache = SqliteCache(constants.IDENTIFIERS_PATH)
 
     def __call__(self, context, config_path, configurable, progress_callback = None):
         """Runs the automagic over the configurable."""
-
-        # Bomb out if we're just the generic interface
-        if self.os is None:
-            return
-
-        # We only need to be called once, so no recursion necessary
-        banners = self.load_banners()
-
-        cacheables = self.find_new_banner_files(banners, self.os)
-
-        new_banners = self.read_new_banners(context, config_path, cacheables, self.symbol_name, self.os,
-                                            progress_callback)
-
-        # 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 doesn'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")
+        self._cache.update(progress_callback)
 
     @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 {operating_system} caches...")
-        for current in range(total):
-            if progress_callback is not None:
-                progress_callback(current * 100 / total, f"Building {operating_system} caches")
-            isf_url = new_urls[current]
-
-            isf = None
-            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(symbol_name).constant_data
-                vollog.log(constants.LOGLEVEL_VV, f"Caching banner {banner} for file {isf_url}")
-
-                bannerlist = banners.get(banner, [])
-                bannerlist.append(isf_url)
-                banners[banner] = bannerlist
-            except exceptions.SymbolError:
-                pass
-            except json.JSONDecodeError:
-                vollog.log(constants.LOGLEVEL_VV, f"Caching file {isf_url} failed due to JSON error")
-            finally:
-                # Get rid of the loaded file, in case it sits in memory
-                if isf:
-                    del isf
-                    gc.collect()
-        return 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
-
-    @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}")
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        """Returns a list of RequirementInterface objects required by this
+        object."""
+        return [requirements.VersionRequirement(name = 'SQLiteCache', component = SqliteCache, version = (1, 0, 0))]
 
 
-class RemoteBannerFormat:
+class RemoteIdentifierFormat:
     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")
+            raise ValueError("Unsupported version for remote identifier list format")
 
     def _verify(self) -> bool:
         version = self._data.get('version', 0)
@@ -188,23 +350,23 @@ class RemoteBannerFormat:
             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(self, identifiers: Dict[bytes, List[str]], operating_system: Optional[str]):
+        raise ValueError("Identifier List version not verified")
 
-    def process_v1(self, banners: Dict[bytes, List[str]], operating_system: Optional[str]):
+    def process_v1(self, identifiers: Optional[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]:
+            for identifier in self._data[operating_system]:
+                binary_identifier = base64.b64decode(identifier)
+                file_list = identifiers.get(binary_identifier, [])
+                for value in self._data[operating_system][identifier]:
                     if value not in file_list:
                         file_list = file_list + [value]
-                    banners[binary_banner] = file_list
+                    identifiers[binary_identifier] = file_list
         if 'additional' in self._data:
             for location in self._data['additional']:
                 try:
-                    subrbf = RemoteBannerFormat(location)
-                    subrbf.process(banners, operating_system)
+                    subrbf = RemoteIdentifierFormat(location)
+                    subrbf.process(identifiers, operating_system)
                 except IOError:
                     vollog.debug(f"Remote file not found: {location}")
-        return banners
+        return identifiers

volatility3/framework/automagic/symbol_finder.py

@@ -3,9 +3,9 @@
 #
 
 import logging
-from typing import Any, Iterable, List, Tuple, Type, Optional, Callable
+from typing import Any, Callable, Iterable, List, Optional, Tuple
 
-from volatility3.framework import interfaces, constants, layers
+from volatility3.framework import constants, interfaces, layers
 from volatility3.framework.automagic import symbol_cache
 from volatility3.framework.configuration import requirements
 from volatility3.framework.layers import scanners
@@ -18,7 +18,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
     priority = 40
 
     banner_config_key: str = "banner"
-    banner_cache: Optional[Type[symbol_cache.SymbolBannerCache]] = None
+    operating_system: Optional[str] = None
     symbol_class: Optional[str] = None
     find_aslr: Optional[Callable] = None
 
@@ -27,14 +27,21 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
         self._requirements: List[Tuple[str, interfaces.configuration.RequirementInterface]] = []
         self._banners: symbol_cache.BannersType = {}
 
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.VersionRequirement(name = 'SQLiteCache',
+                                            component = symbol_cache.SqliteCache,
+                                            version = (1, 0, 0))
+        ]
+
     @property
     def banners(self) -> symbol_cache.BannersType:
         """Creates a cached copy of the results, but only it's been
         requested."""
         if not self._banners:
-            if not self.banner_cache:
-                raise RuntimeError(f"Cache has not been properly defined for {self.__class__.__name__}")
-            self._banners = self.banner_cache.load_banners()
+            cache = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH)
+            self._banners = cache.get_identifier_dictionary(operating_system = self.operating_system)
         return self._banners
 
     def __call__(self,
@@ -103,8 +110,8 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
             vollog.debug(f"Identified banner: {repr(banner)}")
             symbol_files = self.banners.get(banner, None)
             if symbol_files:
-                isf_path = symbol_files[0]
-                vollog.debug(f"Using symbol library: {symbol_files[0]}")
+                isf_path = symbol_files
+                vollog.debug(f"Using symbol library: {symbol_files}")
                 clazz = self.symbol_class
                 # Set the discovered options
                 path_join = interfaces.configuration.path_join
@@ -117,7 +124,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
                 break
             else:
                 if symbol_files:
-                    vollog.debug(f"Symbol library path not found: {symbol_files[0]}")
+                    vollog.debug(f"Symbol library path not found: {symbol_files}")
                     # print("Kernel", banner, hex(banner_offset))
         else:
             vollog.debug("No existing banners found")

volatility3/framework/configuration/requirements.py

@@ -408,13 +408,19 @@ class VersionRequirement(interfaces.configuration.RequirementInterface):
                     config_path: str) -> Dict[str, interfaces.configuration.RequirementInterface]:
         # Mypy doesn't appreciate our classproperty implementation, self._plugin.version has no type
         config_path = interfaces.configuration.path_join(config_path, self.name)
-        if len(self._version) > 0 and self._component.version[0] != self._version[0]:
-            return {config_path: self}
-        if len(self._version) > 1 and self._component.version[1] < self._version[1]:
+        if not self.matches_required(self._version, self._component.version):
             return {config_path: self}
         context.config[interfaces.configuration.path_join(config_path, self.name)] = True
         return {}
 
+    @classmethod
+    def matches_required(cls, required: Tuple[int, ...], version: Tuple[int, int, int]):
+        if len(required) > 0 and version[0] != required[0]:
+            return False
+        if len(required) > 1 and version[1] < required[1]:
+            return False
+        return True
+
 
 class PluginRequirement(VersionRequirement):
 

volatility3/framework/constants/__init__.py

@@ -68,10 +68,13 @@ if sys.platform == 'win32':
 os.makedirs(CACHE_PATH, exist_ok = True)
 
 LINUX_BANNERS_PATH = os.path.join(CACHE_PATH, "linux_banners.cache")
-""""Default location to record information about available linux banners"""
+"""Default location to record information about available linux banners"""
 
 MAC_BANNERS_PATH = os.path.join(CACHE_PATH, "mac_banners.cache")
-""""Default location to record information about available mac banners"""
+"""Default location to record information about available mac banners"""
+
+IDENTIFIERS_PATH = os.path.join(CACHE_PATH, "identifiers.cache")
+"""Default location to record information about available identifiers"""
 
 BUG_URL = "https://github.com/volatilityfoundation/volatility3/issues"
 

volatility3/framework/interfaces/automagic.py

@@ -9,9 +9,9 @@ that a user has not filled.
 """
 import logging
 from abc import ABCMeta
-from typing import Any, List, Optional, Tuple, Union, Type
+from typing import Any, List, Optional, Tuple, Type, Union
 
-from volatility3.framework import interfaces, constants
+from volatility3.framework import constants, interfaces
 from volatility3.framework.configuration import requirements
 
 vollog = logging.getLogger(__name__)
@@ -47,9 +47,10 @@ class AutomagicInterface(interfaces.configuration.ConfigurableInterface, metacla
         super().__init__(context, config_path)
         for requirement in self.get_requirements():
             if not isinstance(requirement, (interfaces.configuration.SimpleTypeRequirement,
-                                            requirements.ChoiceRequirement, requirements.ListRequirement)):
+                                            requirements.ChoiceRequirement, requirements.ListRequirement,
+                                            requirements.VersionRequirement)):
                 raise TypeError(
-                    "Automagic requirements must be a SimpleTypeRequirement, ChoiceRequirement or ListRequirement")
+                    "Automagic requirements must be a SimpleTypeRequirement, ChoiceRequirement, ListRequirement or VersionRequirement")
 
     def __call__(self,
                  context: interfaces.context.ContextInterface,

volatility3/framework/plugins/isfinfo.py

@@ -1,17 +1,16 @@
 # This file is Copyright 2020 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 json
 import logging
 import os
 import pathlib
 import zipfile
-from typing import List, Type, Any, Generator
+from typing import Generator, List
 
 from volatility3 import schemas, symbols
-from volatility3.framework import interfaces, renderers, constants
-from volatility3.framework.automagic import mac, linux, symbol_cache
+from volatility3.framework import constants, interfaces, renderers
+from volatility3.framework.automagic import symbol_cache
 from volatility3.framework.configuration import requirements
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.layers import resources
@@ -23,7 +22,7 @@ class IsfInfo(plugins.PluginInterface):
     """Determines information about the currently available ISF files, or a specific one"""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 0)
+    _version = (2, 0, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -39,7 +38,10 @@ class IsfInfo(plugins.PluginInterface):
             requirements.BooleanRequirement(name = 'validate',
                                             description = 'Validate against schema if possible',
                                             default = False,
-                                            optional = True)
+                                            optional = True),
+            requirements.VersionRequirement(name = 'SQLiteCache',
+                                            component = symbol_cache.SqliteCache,
+                                            version = (1, 0, 0))
         ]
 
     @classmethod
@@ -62,14 +64,6 @@ class IsfInfo(plugins.PluginInterface):
                             if filename.endswith(extension):
                                 yield pathlib.Path(base_name).as_uri()
 
-    def _get_banner(self, clazz: Type[symbol_cache.SymbolBannerCache], data: Any) -> str:
-        """Gets a banner from an ISF file"""
-        banner_symbol = data.get('symbols', {}).get(clazz.symbol_name, {}).get('constant_data',
-                                                                               renderers.NotAvailableValue())
-        if not isinstance(banner_symbol, interfaces.renderers.BaseAbsentValue):
-            banner_symbol = str(base64.b64decode(banner_symbol), encoding = 'latin-1')
-        return banner_symbol
-
     def _generator(self):
         if self.config.get('isf', None) is not None:
             file_list = [self.config['isf']]
@@ -101,7 +95,6 @@ class IsfInfo(plugins.PluginInterface):
         # Process the filtered list
         for entry in filtered_list:
             num_types = num_enums = num_bases = num_symbols = 0
-            windows_info = linux_banner = mac_banner = renderers.NotAvailableValue()
             valid = "Unknown"
             with resources.ResourceAccessor().open(url = entry) as fp:
                 try:
@@ -111,20 +104,20 @@ class IsfInfo(plugins.PluginInterface):
                     num_enums = len(data.get('enums', []))
                     num_bases = len(data.get('base_types', []))
 
-                    linux_banner = self._get_banner(linux.LinuxBannerCache, data)
-                    mac_banner = self._get_banner(mac.MacBannerCache, data)
-                    if not linux_banner and not mac_banner:
-                        windows_info = os.path.splitext(os.path.basename(entry))[0]
+                    identifier_cache = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH)
+                    identifier = identifier_cache.get_identifier(location = entry)
+                    if identifier:
+                        identifier = identifier.decode('utf-8', errors = 'replace')
+                    else:
+                        identifier = renderers.NotAvailableValue()
                     valid = check_valid(data)
                 except (UnicodeDecodeError, json.decoder.JSONDecodeError):
                     vollog.warning(f"Invalid ISF: {entry}")
-            yield (0, (entry, valid, num_bases, num_types, num_symbols, num_enums, windows_info, linux_banner,
-                       mac_banner))
+            yield (0, (entry, valid, num_bases, num_types, num_symbols, num_enums, identifier))
 
     # Try to open the file, load it as JSON, read the data from it
 
     def run(self):
         return renderers.TreeGrid([("URI", str), ("Valid", str),
                                    ("Number of base_types", int), ("Number of types", int), ("Number of symbols", int),
-                                   ("Number of enums", int), ("Windows info", str), ("Linux banner", str),
-                                   ("Mac banner", str)], self._generator())
+                                   ("Number of enums", int), ("Identifying infomration", str)], self._generator())

volatility3/framework/symbols/intermed.py

@@ -202,8 +202,7 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
                     pass
 
             # Finally try looking in zip files
-            zip_path = os.path.join(path, sub_path + ".zip")
-            if os.path.exists(zip_path):
+            for zip_path in pathlib.Path(path).joinpath(sub_path).resolve().rglob(filename + '.zip'):
                 # We have a zipfile, so run through it and look for sub files that match the filename
                 with zipfile.ZipFile(zip_path) as zfile:
                     for name in zfile.namelist():

volatility3/framework/symbols/windows/pdbutil.py

@@ -14,6 +14,8 @@ from urllib import parse, request
 
 from volatility3 import symbols
 from volatility3.framework import constants, contexts, exceptions, interfaces
+from volatility3.framework.automagic import symbol_cache
+from volatility3.framework.configuration import requirements
 from volatility3.framework.configuration.requirements import SymbolTableRequirement
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows import pdbconv
@@ -74,9 +76,15 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
 
         isf_path = None
         # Take the first result of search for the intermediate file
-        for value in intermed.IntermediateSymbolTable.file_symbol_url("windows", filter_string):
+        if not requirements.VersionRequirement.matches_required((1, 0, 0), symbol_cache.SqliteCache.version):
+            vollog.debug(f"Required version of SQLiteCache not found")
+            return None
+
+        value = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH).find_location(
+            symbol_cache.WindowsIdentifier.generate(pdb_name.strip('\x00'), guid.upper(), age), 'windows')
+
+        if value:
             isf_path = value
-            break
         else:
             # If none are found, attempt to download the pdb, convert it and try again
             cls.download_pdb_isf(context, guid.upper(), age, pdb_name, progress_callback)
@@ -336,46 +344,12 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
 
         vollog.debug(f"Found {guid['pdb_name']}: {guid['GUID']}-{guid['age']}")
 
-        module_name = guid["pdb_name"].strip('.pdb')
-
-        symbol_table_name = cls.load_windows_symbol_table(context,
-                                                          guid["GUID"],
-                                                          guid["age"],
-                                                          guid["pdb_name"],
-                                                          "volatility3.framework.symbols.intermed.IntermediateSymbolTable",
-                                                          config_path = config_path)
-
-        new_module_name = None
-        if create_module:
-            new_module = contexts.Module.create(context, module_name, layer_name, offset = guid['mz_offset'],
-                                                symbol_table_name = symbol_table_name)
-            new_module_name = new_module.name
-
-        return new_module_name, symbol_table_name
-
-    @classmethod
-    def module_from_pdb(cls, context: interfaces.context.ContextInterface, config_path: str, layer_name: str,
-                        pdb_name: str, module_offset: int = None, module_size: int = None) -> str:
-        """Creates a module in the specified layer_name based on a pdb name.
-
-        Searches the memory section of the loaded module for its PDB GUID
-        and loads the associated symbol table into the symbol space.
-
-        Args:
-            context: The context to retrieve required elements (layers, symbol tables) from
-            config_path: The config path where to find symbol files
-            layer_name: The name of the layer on which to operate
-            module_offset: This memory dump's module image offset
-            module_size: The size of the module for this dump
-
-        Returns:
-            The name of the constructed and loaded symbol table
-        """
-
-        module_name, _ = cls._modtable_from_pdb(context, config_path, layer_name, pdb_name, module_offset,
-                                                module_size, create_module = True)
-
-        return module_name
+        return cls.load_windows_symbol_table(context,
+                                             guid["GUID"],
+                                             guid["age"],
+                                             guid["pdb_name"],
+                                             "volatility3.framework.symbols.intermed.IntermediateSymbolTable",
+                                             config_path = config_path)
 
 
 class PdbSignatureScanner(interfaces.layers.ScannerInterface):