diff --git a/volatility3/framework/automagic/linux.py b/volatility3/framework/automagic/linux.py index f1d6c91e4..2c152996d 100644 --- a/volatility3/framework/automagic/linux.py +++ b/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'] diff --git a/volatility3/framework/automagic/mac.py b/volatility3/framework/automagic/mac.py index c37aef463..246462878 100644 --- a/volatility3/framework/automagic/mac.py +++ b/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'] diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index 7b6adf9b4..fe717b8be 100644 --- a/volatility3/framework/automagic/symbol_cache.py +++ b/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 diff --git a/volatility3/framework/automagic/symbol_finder.py b/volatility3/framework/automagic/symbol_finder.py index 143abd02e..610ed0e18 100644 --- a/volatility3/framework/automagic/symbol_finder.py +++ b/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") diff --git a/volatility3/framework/configuration/requirements.py b/volatility3/framework/configuration/requirements.py index 4edc6d17c..b31c4767f 100644 --- a/volatility3/framework/configuration/requirements.py +++ b/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): diff --git a/volatility3/framework/constants/__init__.py b/volatility3/framework/constants/__init__.py index f08819f29..322e574e1 100644 --- a/volatility3/framework/constants/__init__.py +++ b/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" diff --git a/volatility3/framework/interfaces/automagic.py b/volatility3/framework/interfaces/automagic.py index c96c9bdbe..713f91da0 100644 --- a/volatility3/framework/interfaces/automagic.py +++ b/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, diff --git a/volatility3/framework/plugins/isfinfo.py b/volatility3/framework/plugins/isfinfo.py index 575f25426..b2960733d 100644 --- a/volatility3/framework/plugins/isfinfo.py +++ b/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()) diff --git a/volatility3/framework/symbols/intermed.py b/volatility3/framework/symbols/intermed.py index a6a7a0fae..1fceb1bcc 100644 --- a/volatility3/framework/symbols/intermed.py +++ b/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(): diff --git a/volatility3/framework/symbols/windows/pdbutil.py b/volatility3/framework/symbols/windows/pdbutil.py index 41037d464..af3741bbe 100644 --- a/volatility3/framework/symbols/windows/pdbutil.py +++ b/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):