Core: Support zip files as plugin paths

volatility3/framework/__init__.py

@@ -5,10 +5,11 @@
 # Check the python version to ensure it's suitable
 import glob
 import sys
+import zipfile
 
 required_python_version = (3, 6, 0)
 if (sys.version_info.major != required_python_version[0] or sys.version_info.minor < required_python_version[1] or
-    (sys.version_info.minor == required_python_version[1] and sys.version_info.micro < required_python_version[2])):
+        (sys.version_info.minor == required_python_version[1] and sys.version_info.micro < required_python_version[2])):
     raise RuntimeError(
         "Volatility framework requires python version {}.{}.{} or greater".format(*required_python_version))
 
@@ -20,6 +21,7 @@ from typing import Any, Dict, Generator, List, Tuple, Type, TypeVar
 
 from volatility3.framework import constants, interfaces
 
+
 # ##
 #
 # SemVer version scheme
@@ -86,7 +88,7 @@ def class_subclasses(cls: Type[T]) -> Generator[Type[T], None, None]:
             yield return_value
 
 
-def import_files(base_module, ignore_errors = False) -> List[str]:
+def import_files(base_module, ignore_errors: bool = False) -> List[str]:
     """Imports all plugins present under plugins module namespace."""
     failures = []
     if not isinstance(base_module.__path__, list):
@@ -94,7 +96,7 @@ def import_files(base_module, ignore_errors = False) -> List[str]:
     vollog.log(constants.LOGLEVEL_VVVV,
                f"Importing from the following paths: {', '.join(base_module.__path__)}")
     for path in base_module.__path__:
-        for root, _, files in os.walk(path, followlinks = True):
+        for root, files in zipwalk(path, followlinks = True):
             # TODO: Figure out how to import pycache files
             if root.endswith("__pycache__"):
                 continue
@@ -115,6 +117,23 @@ def import_files(base_module, ignore_errors = False) -> List[str]:
     return failures
 
 
+def zipwalk(path: str, followlinks: bool = False):
+    """Walks the contents of a zipfile as well as directory"""
+    if zipfile.is_zipfile(path):
+        zip_results = {}
+        with zipfile.ZipFile(path) as archive:
+            for file in archive.filelist:
+                if not file.is_dir():
+                    dirlist = zip_results.get(os.path.dirname(file.filename), [])
+                    dirlist.append(os.path.basename(file.filename))
+                    zip_results[os.path.join(path, os.path.dirname(file.filename))] = dirlist
+        for value in zip_results:
+            yield value, zip_results[value]
+    else:
+        for root, _, files in os.walk(path, followlinks = followlinks):
+            yield root, files
+
+
 def list_plugins() -> Dict[str, Type[interfaces.plugins.PluginInterface]]:
     plugin_list = {}
     for plugin in class_subclasses(interfaces.plugins.PluginInterface):