From 0b6229871fa8f6e854d27607f2b354a0371a6194 Mon Sep 17 00:00:00 2001 From: ikelos Date: Tue, 10 Mar 2026 21:09:05 +0000 Subject: [PATCH] Revert "Only act on local cache symbols under the symbol basepaths" --- .../framework/automagic/symbol_cache.py | 58 +++++-------------- 1 file changed, 14 insertions(+), 44 deletions(-) diff --git a/volatility3/framework/automagic/symbol_cache.py b/volatility3/framework/automagic/symbol_cache.py index ff75e86c6..327575e96 100644 --- a/volatility3/framework/automagic/symbol_cache.py +++ b/volatility3/framework/automagic/symbol_cache.py @@ -239,16 +239,7 @@ class SqliteCache(CacheManagerInterface): results = self._database.cursor().execute(statement, parameters).fetchall() result = None for row in results: - local_filepath = self._get_local_filepath(row["location"]) - if not ( - local_filepath is None - or local_filepath.startswith(tuple(constants.SYMBOL_BASEPATHS)) - ): - vollog.debug( - f"Location {row['location']} found but outside of the registered symbol paths" - ) - else: - result = row["location"] + result = row["location"] return result def get_local_locations(self) -> Generator[str, None, None]: @@ -258,11 +249,7 @@ class SqliteCache(CacheManagerInterface): .fetchall() ) for row in result: - local_filepath = self._get_local_filepath(row["location"]) - if local_filepath and local_filepath.startswith( - tuple(constants.SYMBOL_BASEPATHS) - ): - yield row["location"] + yield row["location"] def is_url_local(self, url: str) -> bool: """Determines whether an url is local or not""" @@ -309,20 +296,6 @@ class SqliteCache(CacheManagerInterface): return row["hash"] return None - def _get_local_filepath( - self, location: str, local_only: bool = True - ) -> Optional[str]: - # See if the file is a local URL type we can handle: - parsed = urllib.parse.urlparse(location) - pathname = location if not local_only else None - if parsed.scheme == "file": - pathname = parsed.path - if parsed.scheme == "jar": - inner_url = urllib.parse.urlparse(parsed.path) - if inner_url.scheme == "file": - pathname = inner_url.path.split("!")[0] - return pathname - 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. @@ -367,7 +340,15 @@ class SqliteCache(CacheManagerInterface): timestamp = stored_timestamp # Default to requiring update # See if the file is a local URL type we can handle: - pathname = self._get_local_filepath(location) + parsed = urllib.parse.urlparse(location) + pathname = None + if parsed.scheme == "file": + pathname = urllib.request.url2pathname(parsed.path) + if parsed.scheme == "jar": + inner_url = urllib.parse.urlparse(parsed.path) + if inner_url.scheme == "file": + pathname = inner_url.path.split("!")[0] + if pathname and os.path.exists(pathname): timestamp = datetime.datetime.fromtimestamp( os.stat(pathname).st_mtime @@ -480,7 +461,7 @@ class SqliteCache(CacheManagerInterface): def get_identifier_dictionary( self, operating_system: Optional[str] = None, local_only: bool = False ) -> Dict[bytes, str]: - output: Dict[bytes, str] = {} + output = {} additions = [] statement = "SELECT location, identifier FROM cache" if local_only: @@ -495,15 +476,7 @@ class SqliteCache(CacheManagerInterface): vollog.debug( f"Duplicate entry for identifier {row['identifier']}: {row['location']} and {output[row['identifier']]}" ) - local_filepath = self._get_local_filepath(row["location"]) - if local_filepath and not local_filepath.startswith( - tuple(constants.SYMBOL_BASEPATHS) - ): - vollog.debug( - f"Location {row['location']} was not in the registered symbol paths and therefore not in the identifier dictionary" - ) - else: - output[row["identifier"]] = row["location"] + output[row["identifier"]] = row["location"] return output def get_identifiers(self, operating_system: Optional[str]) -> List[bytes]: @@ -560,10 +533,7 @@ class SymbolCacheMagic(interfaces.automagic.AutomagicInterface): def __call__(self, context, config_path, configurable, progress_callback=None): """Runs the automagic over the configurable.""" - try: - self._cache.update(progress_callback) - except Exception as excp: - vollog.debug(f"Excption during cache update: {excp}") + self._cache.update(progress_callback) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: