Merge pull request #630 from volatilityfoundation/feature/isf-unification

Feature/isf unification

README.md

@@ -94,6 +94,9 @@ Symbol tables zip files must be placed, as named, into the `volatility3/symbols`
 
 Windows symbols that cannot be found will be queried, downloaded, generated and cached.  Mac and Linux symbol tables must be manually produced by a tool such as [dwarf2json](https://github.com/volatilityfoundation/dwarf2json).
 
+Important: The first run of volatility with new symbol files will require the cache to be updated.  The symbol packs contain a large number of symbol files and so may take some time to update!
+However, this process only needs to be run once on each new symbol file, so assuming the pack stays in the same location will not need to be done again.  Please also note it can be interrupted and next run will restart itself.
+
 Please note: These are representative and are complete up to the point of creation for Windows and Mac.  Due to the ease of compiling Linux kernels and the inability to uniquely distinguish them, an exhaustive set of Linux symbol tables cannot easily be supplied.
 
 ## Documentation

doc/source/symbol-tables.rst

@@ -12,20 +12,22 @@ Volatility will automatically decompress them on use.  It will also cache their
 under the user's home directory, in :file:`.cache/volatility3`, along with other useful data.  The cache directory currently
 cannot be altered.
 
-Symbol table JSON files live, by default, under the :file:`volatility3/symbols`, underneath an operating system directory
-(currently one of :file:`windows`, :file:`mac` or :file:`linux`).  The symbols directory is configurable within the framework and can
-usually be set within the user interface.
+Symbol table JSON files live, by default, under the :file:`volatility3/symbols` directory.  The symbols directory is
+configurable within the framework and can usually be set within the user interface.
 
 These files can also be compressed into ZIP files, which Volatility will process in order to locate symbol files.
-The ZIP file must be named after the appropriate operating system (such as `linux.zip`, `mac.zip` or `windows.zip`).
-Inside the ZIP file, the directory structure should match the uncompressed operating system directory.
+
+Volatility maintains a cache mapping the appropriate identifier for each symbol file against its filename.  This cache
+is updated by automagic called as part of the standard automagic that's run each time a plugin is run.  If a large number of new
+symbols file are detected, this may take some time, but can be safely interrupted and restarted and will not need to run again
+as long as the symbol files stay in the same location.
 
 Windows symbol tables
 ---------------------
 
 For Windows systems, Volatility accepts a string made up of the GUID and Age of the required PDB file.  It then
-searches all files under the configured symbol directories under the windows subdirectory.  Any that match the filename
-pattern of :file:`<pdb-name>/<GUID>-<AGE>.json` (or any compressed variant) will be used.  If such a symbol table cannot be found, then
+searches all files under the configured symbol directories under the windows subdirectory.  Any that contain metadata
+which matches the pdb name and GUID/age (or any compressed variant) will be used.  If such a symbol table cannot be found, then
 the associated PDB file will be downloaded from Microsoft's Symbol Server and converted into the appropriate JSON
 format, and will be saved in the correct location.
 
@@ -41,11 +43,10 @@ or a virtual environment.
 Mac/Linux symbol tables
 -----------------------
 
-For Mac/Linux systems, both use the same mechanism for identification.  JSON files live under the symbol directories,
-under either the :file:`linux` or :file:`mac` directories.  The generated files contain an identifying string (the operating system
+For Mac/Linux systems, both use the same mechanism for identification.  The generated files contain an identifying string (the operating system
 banner), which Volatility's automagic can detect.  Volatility caches the mapping between the strings and the symbol
 tables they come from, meaning the precise file names don't matter and can be organized under any necessary hierarchy
-under the operating system directory.
+under the symbols directory.
 
 Linux and Mac symbol tables can be generated from a DWARF file using a tool called `dwarf2json <https://github.com/volatilityfoundation/dwarf2json>`_.
 Currently a kernel with debugging symbols is the only suitable means for recovering all the information required by
@@ -93,4 +94,4 @@ file, the banners must match exactly (down to the compilation date).
 
   * Copy the `.json` file to the symbols directory into `[symbols directory]/linux`
 
-    * For Mac change `linux` to `mac`
+    * For Mac change `linux` to `mac`

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,21 @@
 # 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, Iterable, List, Optional, Tuple
 
-from volatility3.framework import constants, exceptions, interfaces
+import volatility3.framework
+import volatility3.schemas
+from volatility3 import 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 +25,390 @@ 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 symbol 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) -> Iterable[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]) -> List[bytes]:
+        """Returns all identifiers for a particular operating system"""
+        pass
+
+    def get_location_statistics(self, location: str) -> Optional[Tuple[int, int, int, int]]:
+        """Returns ISF statistics based on the location
+
+        Returns:
+            A tuple of base_types, types, enums, symbols, or None is location not found"""
+
+    def get_hash(self, location: str) -> Optional[str]:
+        """Returns the hash of the JSON from within a location ISF"""
+
+
+class SqliteCache(CacheManagerInterface):
+    _required_framework_version = (2, 0, 0)
+    _version = (1, 0, 0)
+
+    cache_period = '-3 days'
+
+    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) -> sqlite3.Connection:
+        database = sqlite3.connect(path)
+        database.row_factory = sqlite3.Row
+        database.cursor().execute(
+            f'CREATE TABLE IF NOT EXISTS database_info (schema_version INT DEFAULT {constants.CACHE_SQLITE_SCEMA_VERSION})')
+        schema_version = database.cursor().execute('SELECT schema_version FROM database_info').fetchone()
+        if not schema_version:
+            database.cursor().execute(f'INSERT INTO database_info VALUES ({constants.CACHE_SQLITE_SCEMA_VERSION})')
+        elif schema_version['schema_version'] == constants.CACHE_SQLITE_SCEMA_VERSION:
+            # All good, so pass and move on
+            pass
+        else:
+            vollog.info(f"Previous cache schema version found: {schema_version['schema_version']}")
+            # TODO: Implement code if the schema changes
+            # Current this should never happen so we start over again
+            database.close()
+            os.unlink(path)
+            return self._connect_storage(path)
+        database.cursor().execute(
+            'CREATE TABLE IF NOT EXISTS cache (location TEXT UNIQUE NOT NULL, identifier TEXT, operating_system TEXT, hash TEXT,'
+            'stats_base_types INT DEFAULT 0, stats_types INT DEFAULT 0, stats_enums INT DEFAULT 0, stats_symbols INT DEFAULT 0, local BOOL, cached DATETIME)')
+        database.commit()
+        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 get_location_statistics(self, location: str) -> Optional[Tuple[int, int, int, int]]:
+        results = self._database.cursor().execute(
+            'SELECT stats_base_types, stats_types, stats_enums, stats_symbols FROM cache WHERE location = ?',
+            (location,)).fetchall()
+        for row in results:
+            return row['stats_base_types'], row['stats_types'], row['stats_enums'], row['stats_symbols']
+        return None
+
+    def get_hash(self, location: str) -> Optional[str]:
+        results = self._database.cursor().execute('SELECT hash FROM cache WHERE location = ?',
+                                                  (location,)).fetchall()
+        for row in results:
+            return row['hash']
+
+    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 "
+                                                     f"AND cached < date('now', '{self.cache_period}');")
+            for row in result:
+                if row['location'] in files_to_timestamp:
+                    cache_update.add(row['location'])
+
+        idextractors = list(volatility3.framework.class_subclasses(IdentifierProcessor))
+
+        # New or not recently updated
+
+        files_to_process = new_locations.union(cache_update)
+        number_files_to_process = len(files_to_process)
+        cursor = self._database.cursor()
+        try:
+            for counter, location in enumerate(files_to_process):
+                # Open location
+                progress_callback(counter * 100 / number_files_to_process,
+                                  f"Updating caches for {number_files_to_process} files...")
+                try:
+                    with resources.ResourceAccessor().open(location) as fp:
+                        json_obj = json.load(fp)
+                        hash = schemas.create_json_hash(json_obj)
+                        identifier = None
+
+                        # Get stats
+                        stats_base_types = len(json_obj.get('base_types', {}))
+                        stats_types = len(json_obj.get('types', {}))
+                        stats_enums = len(json_obj.get('enums', {}))
+                        stats_symbols = len(json_obj.get('symbols', {}))
+
+                        operating_system = None
+                        for idextractor in idextractors:
+                            identifier = idextractor.get_identifier(json_obj)
+                            if identifier is not None:
+                                operating_system = idextractor.operating_system
+                                break
+
+                        # We don't try to validate schemas here, we do that on first use
+                        # Store in database
+                        cursor.execute(
+                            "INSERT OR REPLACE INTO cache (location, identifier, operating_system, hash,"
+                            "stats_base_types, stats_types, stats_enums, stats_symbols, "
+                            "local, cached) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))",
+                            (
+                                location,
+                                identifier,
+                                operating_system,
+                                hash,
+                                stats_base_types,
+                                stats_types,
+                                stats_enums,
+                                stats_symbols,
+                                self.is_url_local(location)
+                            ))
+                        if identifier is not None:
+                            vollog.log(constants.LOGLEVEL_VV, f"Identified {location} as {identifier}")
+                        else:
+                            vollog.log(constants.LOGLEVEL_VVVV, f"No identifier found for {location}")
+                except Exception as excp:
+                    vollog.log(constants.LOGLEVEL_VVVV, excp)
+        finally:
+            self._database.commit()
+
+        # Remote Entries
+
+        if not constants.OFFLINE and constants.REMOTE_ISF_URL:
+            progress_callback(0, 'Reading remote ISF list')
+            cursor = self._database.cursor()
+            cursor.execute(
+                f"SELECT cached FROM cache WHERE remote = True and cached < datetime('now', {self.cache_period})")
+            remote_identifiers = RemoteIdentifierFormat(constants.REMOTE_ISF_URL)
+            progress_callback(50, 'Reading remote ISF list')
+            for operating_system in constants.OS_CATEGORIES:
+                identifiers = remote_identifiers.process({}, operating_system = operating_system)
+                for identifier, location in identifiers:
+                    cursor.execute(
+                        "INSERT OR REPLACE INTO cache(identifier, location, operating_system, local, cached) VALUES (?, ?, ?, ?, datetime('now'))",
+                        (location, identifier, operating_system, False)
+                    )
+            progress_callback(100, 'Reading remote ISF list')
+            self._database.commit()
+
+        # Missing entries
+
+        if missing_locations:
+            self._database.cursor().execute(
+                f"DELETE FROM cache WHERE location IN ({','.join(['?'] * len(missing_locations))})", [x for x in missing_locations])
+            self._database.commit()
+
+    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]) -> List[bytes]:
+        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 +417,22 @@ 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]) -> Generator[
+        Tuple[bytes, str], None, None]:
+        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]) -> Generator[
+        Tuple[bytes, str], None, None]:
         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
+            for identifier in self._data[operating_system]:
+                binary_identifier = base64.b64decode(identifier)
+                for value in self._data[operating_system][identifier]:
+                    yield binary_identifier, value
         if 'additional' in self._data:
             for location in self._data['additional']:
                 try:
-                    subrbf = RemoteBannerFormat(location)
-                    subrbf.process(banners, operating_system)
+                    subrbf = RemoteIdentifierFormat(location)
+                    yield from 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]) -> bool:
+        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,16 @@ 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"""
+
+CACHE_SQLITE_SCEMA_VERSION = 1
+"""Version for the sqlite3 cache schema"""
 
 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,6 +38,13 @@ class IsfInfo(plugins.PluginInterface):
             requirements.BooleanRequirement(name = 'validate',
                                             description = 'Validate against schema if possible',
                                             default = False,
+                                            optional = True),
+            requirements.VersionRequirement(name = 'SQLiteCache',
+                                            component = symbol_cache.SqliteCache,
+                                            version = (1, 0, 0)),
+            requirements.BooleanRequirement(name = 'live',
+                                            description = 'Traverse all files, rather than use the cache',
+                                            default = False,
                                             optional = True)
         ]
 
@@ -62,14 +68,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']]
@@ -98,33 +96,52 @@ class IsfInfo(plugins.PluginInterface):
             def check_valid(data):
                 return "Unknown"
 
-        # 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:
-                    data = json.load(fp)
-                    num_symbols = len(data.get('symbols', []))
-                    num_types = len(data.get('user_types', []))
-                    num_enums = len(data.get('enums', []))
-                    num_bases = len(data.get('base_types', []))
+        if self.config['live']:
+            # Process the filtered list
+            for entry in filtered_list:
+                num_types = num_enums = num_bases = num_symbols = 0
+                valid = "Unknown"
+                with resources.ResourceAccessor().open(url = entry) as fp:
+                    try:
+                        data = json.load(fp)
+                        num_symbols = len(data.get('symbols', []))
+                        num_types = len(data.get('user_types', []))
+                        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]
-                    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))
+                        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, identifier))
+        else:
+            cache = symbol_cache.SqliteCache(constants.IDENTIFIERS_PATH)
+            valid = 'Unknown'
+            for identifier, location in cache.get_identifier_dictionary().items():
+                num_bases, num_types, num_enums, num_symbols = cache.get_location_statistics(location)
+                if identifier:
+                    json_hash = cache.get_hash(location)
+                    if json_hash and json_hash in schemas.cached_validations:
+                        valid = 'True (cached)'
+                    if self.config['validate']:
+                        # Even if we're not live, if we've been explicitly asked to validate, then do-so
+                        with resources.ResourceAccessor().open(url = location) as fp:
+                            try:
+                                data = json.load(fp)
+                                valid = check_valid(data)
+                            except (UnicodeDecodeError, json.decoder.JSONDecodeError):
+                                vollog.warning(f"Invalid ISF: {location}")
+
+                    yield (0, (location, valid, num_bases, num_types, num_symbols, num_enums, str(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 information", 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/pdbconv.py

@@ -521,7 +521,7 @@ class PdbReader:
 
         self.metadata['windows']['pdb'] = {
             "GUID": self.convert_bytes_to_guid(pdb_info.GUID),
-            "age": pdb_info.age,
+            "age": self._dbiheader.age,
             "database": self._database_name or 'unknown.pdb',
             "machine_type": self._dbiheader.machine
         }

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

volatility3/schemas/__init__.py

@@ -6,7 +6,7 @@ import hashlib
 import json
 import logging
 import os
-from typing import Set, Any, Dict
+from typing import Any, Dict, Optional, Set
 
 from volatility3.framework import constants
 
@@ -51,9 +51,21 @@ def validate(input: Dict[str, Any], use_cache: bool = True) -> bool:
     return valid(input, schema, use_cache)
 
 
-def create_json_hash(input: Dict[str, Any], schema: Dict[str, Any]) -> str:
+def create_json_hash(input: Dict[str, Any], schema: Optional[Dict[str, Any]] = None) -> Optional[str]:
     """Constructs the hash of the input and schema to create a unique
     identifier for a particular JSON file."""
+    if schema is None:
+        format = input.get('metadata', {}).get('format', None)
+        if not format:
+            vollog.debug("No schema format defined")
+            return None
+        basepath = os.path.abspath(os.path.dirname(__file__))
+        schema_path = os.path.join(basepath, 'schema-' + format + '.json')
+        if not os.path.exists(schema_path):
+            vollog.debug(f"Schema for format not found: {schema_path}")
+            return None
+        with open(schema_path, 'r') as s:
+            schema = json.load(s)
     return hashlib.sha1(bytes(json.dumps((input, schema), sort_keys = True), 'utf-8')).hexdigest()