Plugins: Update isfinfo to use the cache unless --live

volatility3/framework/automagic/symbol_cache.py

@@ -104,7 +104,7 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
         """Returns the location of the symbol file given the identifier
 
         Args:
-            identifier: string that uniquely identifies a particular symbolt table
+            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:
@@ -144,6 +144,18 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
         """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_verified(self, location: str) -> bool:
+        """Returns whether a location ISF has been verified against its schema"""
+
+    def set_verified(self, location: str, state: bool = True) -> None:
+        """Sets the verified state of a location based on whether it has been successfully verified against its schema"""
+
 
 class SqliteCache(CacheManagerInterface):
     _required_framework_version = (2, 0, 0)
@@ -163,7 +175,23 @@ class SqliteCache(CacheManagerInterface):
         database = sqlite3.connect(path)
         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)')
+            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, verified BOOL DEFAULT False,'
+            '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
 
@@ -207,6 +235,25 @@ class SqliteCache(CacheManagerInterface):
             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_verified(self, location: str) -> bool:
+        results = self._database.cursor().execute('SELECT verified FROM cache WHERE location = ?',
+                                                  (location,)).fetchall()
+        for row in results:
+            return row['verified']
+        return False
+
+    def set_verified(self, location: str, state: bool = True) -> None:
+        self._database.cursor().execute('UPDATE cache (verified) VALUES (?) WHERE location = ?',
+                                        (state, location,))
+
     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.
@@ -245,32 +292,39 @@ class SqliteCache(CacheManagerInterface):
                     with resources.ResourceAccessor().open(location) as fp:
                         json_obj = json.load(fp)
                         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)
-                            operating_system = idextractor.operating_system
                             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, "
+                            "stats_base_types, stats_types, stats_enums, stats_symbols, "
+                            "local, cached) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))",
+                            (
+                                location,
+                                identifier,
+                                operating_system,
+                                stats_base_types,
+                                stats_types,
+                                stats_enums,
+                                stats_symbols,
+                                self.is_url_local(location)
+                            ))
                         if identifier is not None:
-                            # 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, 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:
-                            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)

volatility3/framework/constants/__init__.py

@@ -76,6 +76,9 @@ MAC_BANNERS_PATH = os.path.join(CACHE_PATH, "mac_banners.cache")
 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"
 
 ProgressCallback = Optional[Callable[[float, str], None]]

volatility3/framework/plugins/isfinfo.py

@@ -41,7 +41,11 @@ class IsfInfo(plugins.PluginInterface):
                                             optional = True),
             requirements.VersionRequirement(name = 'SQLiteCache',
                                             component = symbol_cache.SqliteCache,
-                                            version = (1, 0, 0))
+                                            version = (1, 0, 0)),
+            requirements.BooleanRequirement(name = 'live',
+                                            description = 'Traverse all files, rather than use the cache',
+                                            default = False,
+                                            optional = True)
         ]
 
     @classmethod
@@ -92,28 +96,36 @@ 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
-            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', []))
 
-                    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))
+                        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:
+                    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