Revert "Only act on local cache symbols under the symbol basepaths"

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