From 1f02fea5d10be5c193f2b100bb18c973335504be Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Sat, 26 Feb 2022 23:40:47 +0000 Subject: [PATCH] Automagic: Change database to store ISF hash instead of verified state --- .../framework/automagic/symbol_cache.py | 27 ++++++++----------- volatility3/framework/plugins/isfinfo.py | 12 +++++++++ volatility3/schemas/__init__.py | 16 +++++++++-- 3 files changed, 37 insertions(+), 18 deletions(-) diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index c09904713..77bc46265 100644 --- a/volatility3/framework/automagic/symbol_cache.py +++ b/volatility3/framework/automagic/symbol_cache.py @@ -14,6 +14,7 @@ from typing import Dict, Generator, List, Optional, Tuple 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 @@ -150,11 +151,8 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface): 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""" + def get_hash(self, location: str) -> bool: + """Returns the hash of the JSON from within a location ISF""" class SqliteCache(CacheManagerInterface): @@ -190,7 +188,7 @@ class SqliteCache(CacheManagerInterface): 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,' + '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 @@ -243,16 +241,11 @@ class SqliteCache(CacheManagerInterface): 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 = ?', + 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['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,)) + return row['hash'] def update(self, progress_callback = None): """Locates all files under the symbol directories. Updates the cache with additions, modifications and removals. @@ -291,6 +284,7 @@ class SqliteCache(CacheManagerInterface): try: with resources.ResourceAccessor().open(location) as fp: json_obj = json.load(fp) + hash = schemas.create_json_hash(json_obj) identifier = None # Get stats @@ -309,13 +303,14 @@ class SqliteCache(CacheManagerInterface): # 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, " + "INSERT OR REPLACE INTO cache (location, identifier, operating_system, hash," "stats_base_types, stats_types, stats_enums, stats_symbols, " - "local, cached) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))", + "local, cached) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))", ( location, identifier, operating_system, + hash, stats_base_types, stats_types, stats_enums, diff --git a/volatility3/framework/plugins/isfinfo.py b/volatility3/framework/plugins/isfinfo.py index b94cfd69a..af095b69d 100644 --- a/volatility3/framework/plugins/isfinfo.py +++ b/volatility3/framework/plugins/isfinfo.py @@ -125,6 +125,18 @@ class IsfInfo(plugins.PluginInterface): 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 diff --git a/volatility3/schemas/__init__.py b/volatility3/schemas/__init__.py index 65329a4f5..8666680b3 100644 --- a/volatility3/schemas/__init__.py +++ b/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()