Merge branch 'volatilityfoundation:develop' into fix/workflow

development/mac-kdk/parse_pbzx2.py

@@ -17,61 +17,54 @@ def seekread(f, offset = None, length = 0, relative = True):
         f.seek(offset, [0, 1, 2][relative])
     if length:
         return f.read(length)
+    return None
 
 
 def parse_pbzx(pbzx_path):
     section = 0
     xar_out_path = '%s.part%02d.cpio.xz' % (pbzx_path, section)
-    f = open(pbzx_path, 'rb')
-    # pbzx = f.read()
-    # f.close()
-    magic = seekread(f, length = 4)
-    if magic != 'pbzx':
-        raise RuntimeError("Error: Not a pbzx file")
-    # Read 8 bytes for initial flags
-    flags = seekread(f, length = 8)
-    # Interpret the flags as a 64-bit big-endian unsigned int
-    flags = struct.unpack('>Q', flags)[0]
-    xar_f = open(xar_out_path, 'wb')
-    while flags & (1 << 24):
-        # Read in more flags
+    with open(pbzx_path, 'rb') as f:
+        # pbzx = f.read()
+        # f.close()
+        magic = seekread(f, length = 4)
+        if magic != 'pbzx':
+            raise RuntimeError("Error: Not a pbzx file")
+        # Read 8 bytes for initial flags
         flags = seekread(f, length = 8)
+        # Interpret the flags as a 64-bit big-endian unsigned int
         flags = struct.unpack('>Q', flags)[0]
-        # Read in length
-        f_length = seekread(f, length = 8)
-        f_length = struct.unpack('>Q', f_length)[0]
-        xzmagic = seekread(f, length = 6)
-        if xzmagic != '\xfd7zXZ\x00':
-            # This isn't xz content, this is actually _raw decompressed cpio_ chunk of 16MB in size...
-            # Let's back up ...
-            seekread(f, offset = -6, length = 0)
-            # ... and split it out ...
-            f_content = seekread(f, length = f_length)
-            section += 1
-            decomp_out = '%s.part%02d.cpio' % (pbzx_path, section)
-            g = open(decomp_out, 'wb')
-            g.write(f_content)
-            g.close()
-            # Now to start the next section, which should hopefully be .xz (we'll just assume it is ...)
-            xar_f.close()
-            section += 1
-            new_out = '%s.part%02d.cpio.xz' % (pbzx_path, section)
-            xar_f = open(new_out, 'wb')
-        else:
-            f_length -= 6
-            # This part needs buffering
-            f_content = seekread(f, length = f_length)
-            tail = seekread(f, offset = -2, length = 2)
-            xar_f.write(xzmagic)
-            xar_f.write(f_content)
-            if tail != 'YZ':
-                xar_f.close()
-                raise RuntimeError("Error: Footer is not xar file footer")
-    try:
-        f.close()
-        xar_f.close()
-    except IOError:
-        pass
+        while flags & (1 << 24):
+            with open(xar_out_path, 'wb') as xar_f:
+                xar_f.seek(0, os.SEEK_END)
+                # Read in more flags
+                flags = seekread(f, length = 8)
+                flags = struct.unpack('>Q', flags)[0]
+                # Read in length
+                f_length = seekread(f, length = 8)
+                f_length = struct.unpack('>Q', f_length)[0]
+                xzmagic = seekread(f, length = 6)
+                if xzmagic != '\xfd7zXZ\x00':
+                    # This isn't xz content, this is actually _raw decompressed cpio_ chunk of 16MB in size...
+                    # Let's back up ...
+                    seekread(f, offset = -6, length = 0)
+                    # ... and split it out ...
+                    f_content = seekread(f, length = f_length)
+                    section += 1
+                    decomp_out = '%s.part%02d.cpio' % (pbzx_path, section)
+                    with open(decomp_out, 'wb') as g:
+                        g.write(f_content)
+                    # Now to start the next section, which should hopefully be .xz (we'll just assume it is ...)
+                    section += 1
+                    xar_out_path = '%s.part%02d.cpio.xz' % (pbzx_path, section)
+                else:
+                    f_length -= 6
+                    # This part needs buffering
+                    f_content = seekread(f, length = f_length)
+                    tail = seekread(f, offset = -2, length = 2)
+                    xar_f.write(xzmagic)
+                    xar_f.write(f_content)
+                    if tail != 'YZ':
+                        raise RuntimeError("Error: Footer is not xar file footer")
 
 
 def main():

doc/source/conf.py

@@ -121,6 +121,7 @@ try:
 
     extensions.append('sphinx_autodoc_typehints')
 except ImportError:
+    # If the autodoc typehints extension isn't available, carry on regardless
     pass
 
 # Add any paths that contain templates here, relative to this directory.

requirements-dev.txt

@@ -16,7 +16,7 @@ pycryptodome
 
 # This can improve error messages regarding improperly configured ISF files,
 # but is only recommended for development
-# jsonschema>=2.3.0
+jsonschema>=2.3.0
 
 # This is required for memory acquisition via leechcore/pcileech.
 leechcorepyc>=2.4.0

volatility3/cli/volshell/__init__.py

@@ -36,7 +36,6 @@ class VolShell(cli.CommandLine):
 
     def __init__(self):
         super().__init__()
-        self.output_dir = None
 
     def run(self):
         """Executes the command line module, taking the system arguments,

volatility3/cli/volshell/generic.py

@@ -324,11 +324,11 @@ class Volshell(interfaces.plugins.PluginInterface):
                           " " * (longest_member - len_member), "  ", member_type.vol.type_name)
 
     @classmethod
-    def _display_value(self, value: Any) -> str:
+    def _display_value(cls, value: Any) -> str:
         if isinstance(value, objects.PrimitiveObject):
             return repr(value)
         elif isinstance(value, objects.Array):
-            return repr([self._display_value(val) for val in value])
+            return repr([cls._display_value(val) for val in value])
         else:
             return hex(value.vol.offset)
 
@@ -390,8 +390,8 @@ class Volshell(interfaces.plugins.PluginInterface):
             location = "file:" + request.pathname2url(location)
         print(f"Running code from {location}\n")
         accessor = resources.ResourceAccessor()
-        with io.TextIOWrapper(accessor.open(url = location), encoding = 'utf-8') as fp:
-            self.__console.runsource(fp.read(), symbol = 'exec')
+        with accessor.open(url = location) as fp:
+            self.__console.runsource(io.TextIOWrapper(fp.read(), encoding = 'utf-8'), symbol = 'exec')
         print("\nCode complete")
 
     def load_file(self, location: str):

volatility3/framework/automagic/pdbscan.py

@@ -181,6 +181,7 @@ class KernelPDBScanner(interfaces.automagic.AutomagicInterface):
                         hex(kvo)))
             except exceptions.InvalidAddressException:
                 vollog.debug(f"Potential kernel_virtual_offset caused a page fault: {hex(kvo)}")
+            return None
 
         vollog.debug("Kernel base determination - testing fixed base address")
         return self._method_layer_pdb_scan(context, vlayer, test_physical_kernel, False, True, progress_callback)

volatility3/framework/automagic/symbol_cache.py

@@ -223,8 +223,7 @@ class SqliteCache(CacheManagerInterface):
     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
+        return parsed.scheme in ['file', 'jar']
 
     def get_identifier(self, location: str) -> Optional[bytes]:
         results = self._database.cursor().execute('SELECT identifier FROM cache WHERE location = ?',
@@ -246,6 +245,7 @@ class SqliteCache(CacheManagerInterface):
                                                   (location,)).fetchall()
         for row in results:
             return row['hash']
+        return None
 
     def update(self, progress_callback = None):
         """Locates all files under the symbol directories.  Updates the cache with additions, modifications and removals.

volatility3/framework/interfaces/automagic.py

@@ -113,7 +113,7 @@ class StackerLayerInterface(metaclass = ABCMeta):
     """The list operating systems/first-level plugin hierarchy that should exclude this stacker"""
 
     @classmethod
-    def stack(self,
+    def stack(cls,
               context: interfaces.context.ContextInterface,
               layer_name: str,
               progress_callback: constants.ProgressCallback = None) -> Optional[interfaces.layers.DataLayerInterface]:

volatility3/framework/layers/resources.py

@@ -31,6 +31,7 @@ try:
     # Import so that the handler is found by the framework.class_subclasses callc
     import smb.SMBHandler  # lgtm [py/unused-import]
 except ImportError:
+    # If we fail to import this, it means that SMB handling won't be available
     pass
 
 vollog = logging.getLogger(__name__)

volatility3/framework/layers/vmware.py

@@ -154,7 +154,8 @@ class VmwareStacker(interfaces.automagic.StackerLayerInterface):
 
             vmss_success = False
             with contextlib.suppress(IOError):
-                _ = resources.ResourceAccessor().open(vmss).read(10)
+                with resources.ResourceAccessor().open(vmss) as fp:
+                    _ = fp.read(10)
                 context.config[interfaces.configuration.path_join(current_config_path, "location")] = vmss
                 context.layers.add_layer(physical.FileLayer(context, current_config_path, current_layer_name))
                 vmss_success = True

volatility3/framework/objects/__init__.py

@@ -747,10 +747,8 @@ class AggregateType(interfaces.objects.ObjectInterface):
             if isinstance(cls, agg_type):
                 agg_name = agg_type.__name__
 
-        assert isinstance(members, collections.abc.Mapping)
-        f"{agg_name} members parameter must be a mapping: {type(members)}"
-        assert all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()])
-        f"{agg_name} members must be a tuple of relative_offsets and templates"
+        assert isinstance(members, collections.abc.Mapping), f"{agg_name} members parameter must be a mapping: {type(members)}"
+        assert all([(isinstance(member, tuple) and len(member) == 2) for member in members.values()]), f"{agg_name} members must be a tuple of relative_offsets and templates"
 
     def member(self, attr: str = 'member') -> object:
         """Specifically named method for retrieving members."""

volatility3/framework/plugins/linux/check_modules.py

@@ -29,7 +29,7 @@ class Check_modules(plugins.PluginInterface):
         ]
 
     @classmethod
-    def get_kset_modules(self, context: interfaces.context.ContextInterface, vmlinux_name: str):
+    def get_kset_modules(cls, context: interfaces.context.ContextInterface, vmlinux_name: str):
 
         vmlinux = context.modules[vmlinux_name]
 

volatility3/framework/plugins/mac/lsmod.py

@@ -46,14 +46,14 @@ class Lsmod(plugins.PluginInterface):
         try:
             kmod = kmod_ptr.dereference().cast("kmod_info")
         except exceptions.InvalidAddressException:
-            return []
+            return # Generation finished
 
         yield kmod
 
         try:
             kmod = kmod.next
         except exceptions.InvalidAddressException:
-            return []
+            return # Generation finished
 
         seen: Set = set()
 
@@ -74,6 +74,7 @@ class Lsmod(plugins.PluginInterface):
                 kmod = kmod.next
             except exceptions.InvalidAddressException:
                 return
+        return # Generation finished
 
     def _generator(self):
         for module in self.list_modules(self.context, self.config['kernel']):

volatility3/framework/plugins/windows/cachedump.py

@@ -83,6 +83,13 @@ class Cachedump(interfaces.plugins.PluginInterface):
         return (username, domain, domain_name, hashh)
 
     def _generator(self, syshive, sechive):
+        if not syshive or not sechive:
+            if syshive is None:
+                vollog.warning('Unable to locate SYSTEM hive')
+            if sechive is None:
+                vollog.warning('Unable to locate SECURITY hive')
+            return
+
         bootkey = hashdump.Hashdump.get_bootkey(syshive)
         if not bootkey:
             vollog.warning('Unable to find bootkey')
@@ -142,12 +149,5 @@ class Cachedump(interfaces.plugins.PluginInterface):
             if hive.get_name().split('\\')[-1].upper() == 'SECURITY':
                 sechive = hive
 
-        if syshive is None or sechive is None:
-            if syshive is None:
-                vollog.warning('Unable to locate SYSTEM hive')
-            if sechive is None:
-                vollog.warning('Unable to locate SECURITY hive')
-            return
-
         return renderers.TreeGrid([("Username", str), ("Domain", str), ("Domain name", str), ('Hash', bytes)],
                                   self._generator(syshive, sechive))

volatility3/framework/plugins/windows/netstat.py

@@ -42,7 +42,7 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         ]
 
     @classmethod
-    def _decode_pointer(self, value):
+    def _decode_pointer(cls, value):
         """Copied from `windows.handles`.
 
         Windows encodes pointers to objects and decodes them on the fly
@@ -427,6 +427,8 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                                                                            self.config_path)
 
         tcpip_module = self.get_tcpip_module(self.context, kernel.layer_name, kernel.symbol_table_name)
+        if not tcpip_module:
+            vollog.error("Unable to locate symbols for the memory image's tcpip module")
 
         try:
             tcpip_symbol_table = pdbutil.PDBUtility.symbol_table_from_pdb(

volatility3/framework/plugins/windows/pslist.py

@@ -62,7 +62,9 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         """
 
         file_handle = None
+        proc_id = 'Invalid process object'
         try:
+            proc_id = proc.UniqueProcessId
             proc_layer_name = proc.add_process_layer()
             peb = context.object(kernel_table_name + constants.BANG + "_PEB",
                                  layer_name = proc_layer_name,
@@ -76,7 +78,7 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                 file_handle.seek(offset)
                 file_handle.write(data)
         except Exception as excp:
-            vollog.debug(f"Unable to dump PE with pid {proc.UniqueProcessId}: {excp}")
+            vollog.debug(f"Unable to dump PE with pid {proc_id}: {excp}")
 
         return file_handle
 

volatility3/framework/plugins/windows/psscan.py

@@ -4,7 +4,7 @@
 
 import datetime
 import logging
-from typing import Iterable, Callable, Tuple
+from typing import Iterable, Callable, Optional, Tuple
 
 from volatility3.framework import renderers, interfaces, layers, exceptions
 from volatility3.framework.configuration import requirements
@@ -78,7 +78,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                                       layer_name: str,
                                       symbol_table: str,
                                       proc: interfaces.objects.ObjectInterface) -> \
-            Iterable[interfaces.objects.ObjectInterface]:
+            Optional[interfaces.objects.ObjectInterface]:
         """ Returns a virtual process from a physical addressed one
 
         Args:
@@ -124,6 +124,7 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             if virtual_process and \
                     proc.vol.offset == ph_offset:
                 return virtual_process
+        return None
 
     @classmethod
     def get_osversion(cls, context: interfaces.context.ContextInterface, layer_name: str,

volatility3/framework/plugins/windows/registry/userassist.py

@@ -33,7 +33,11 @@ class UserAssist(interfaces.plugins.PluginInterface):
         self._reg_table_name = None
         self._win7 = None
         # taken from http://msdn.microsoft.com/en-us/library/dd378457%28v=vs.85%29.aspx
-        self._folder_guids = json.load(open(os.path.join(os.path.dirname(__file__), "userassist.json"), "rb"))
+        try:
+            with open(os.path.join(os.path.dirname(__file__), "userassist.json"), "rb") as fp:
+                self._folder_guids = json.load(fp)
+        except IOError:
+            vollog.error("Usersassist data file not found")
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:

volatility3/framework/symbols/intermed.py

@@ -102,10 +102,9 @@ class IntermediateSymbolTable(interfaces.symbols.SymbolTableInterface):
         # Check there are no obvious errors
         # Open the file and test the version
         self._versions = dict([(x.version, x) for x in class_subclasses(ISFormatTable)])
-        fp = resources.ResourceAccessor().open(isf_url)
-        reader = codecs.getreader("utf-8")
-        json_object = json.load(reader(fp))  # type: ignore
-        fp.close()
+        with resources.ResourceAccessor().open(isf_url) as fp:
+            reader = codecs.getreader("utf-8")
+            json_object = json.load(reader(fp))  # type: ignore
 
         # Validation is expensive, but we cache to store the hashes of successfully validated json objects
         if validate and not schemas.validate(json_object):

volatility3/framework/symbols/linux/extensions/__init__.py

@@ -128,6 +128,7 @@ class module(generic.GenericIntelProcess):
             sym_addr = sym.st_value
             if wanted_sym_name == sym_name:
                 return sym_addr
+        return None
 
     @property
     def section_symtab(self):

volatility3/framework/symbols/metadata.py

@@ -2,7 +2,7 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Union
 
 from volatility3.framework import interfaces
 
@@ -11,7 +11,7 @@ class WindowsMetadata(interfaces.symbols.MetadataInterface):
     """Class to handle the metadata from a Windows symbol table."""
 
     @property
-    def pe_version(self) -> Optional[Tuple]:
+    def pe_version(self) -> Optional[Union[Tuple[int, int, int], Tuple[int, int, int, int]]]:
         build = self._json_data.get('pe', {}).get('build', None)
         revision = self._json_data.get('pe', {}).get('revision', None)
         minor = self._json_data.get('pe', {}).get('minor', None)

volatility3/framework/symbols/windows/extensions/__init__.py

@@ -719,7 +719,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
             envars = context.layers[process_space].read(block, block_size).decode("utf-16-le",
                                                                                   errors = 'replace').split('\x00')[:-1]
         except exceptions.InvalidAddressException:
-            return renderers.UnreadableValue()
+            return # Generation finished
 
         for envar in envars:
             split_index = envar.find('=')
@@ -729,6 +729,7 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
             # Exclude parse problem with some types of env
             if env and var:
                 yield env, var
+        return # Generation finished
 
 
 class LIST_ENTRY(objects.StructType, collections.abc.Iterable):

volatility3/framework/symbols/windows/pdbconv.py

@@ -926,14 +926,16 @@ class PdbRetreiver:
                 try:
                     vollog.debug(f"Attempting to retrieve {url + suffix}")
                     # We have to cache this because the file is opened by a layer and we can't control whether that caches
-                    result = resources.ResourceAccessor(progress_callback).open(url + suffix)
+                    with resources.ResourceAccessor(progress_callback).open(url + suffix) as fp:
+                        fp.read(10)
+                        result = True
                 except (error.HTTPError, error.URLError) as excp:
                     vollog.debug(f"Failed with {excp}")
                 if result:
                     break
         if progress_callback is not None:
             progress_callback(100, f"Downloading {url + suffix}")
-        if result is None:
+        if not result:
             return None
         return url + suffix
 

volatility3/framework/symbols/windows/pdbutil.py

@@ -359,7 +359,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
 
         vollog.debug(f"Found {guid['pdb_name']}: {guid['GUID']}-{guid['age']}")
 
-        module_name = guid["pdb_name"].strip('.pdb')
+        module_name = guid["pdb_name"].replace('.pdb', '')
 
         symbol_table_name = cls.load_windows_symbol_table(context,
                                                           guid["GUID"],

volatility3/plugins/windows/registry/certificates.py

@@ -3,7 +3,7 @@ import logging
 import struct
 from typing import List, Iterator, Optional, Tuple, Type
 
-from volatility3.framework import constants, exceptions, interfaces, renderers
+from volatility3.framework import exceptions, interfaces, renderers
 from volatility3.framework.configuration import requirements
 from volatility3.framework.symbols.windows.extensions.registry import RegValueTypes
 from volatility3.plugins.windows.registry import hivelist, printkey
@@ -46,14 +46,13 @@ class Certificates(interfaces.plugins.PluginInterface):
                         open_method: Type[interfaces.plugins.FileHandlerInterface]) -> \
                         Optional[interfaces.plugins.FileHandlerInterface]:
         try:
-            if not isinstance(certificate_data, interfaces.renderers.BaseAbsentValue):
-                dump_name = "{}-{}-{}.crt".format(hive_offset, reg_section, key_hash)
-                file_handle = open_method(dump_name)
-                file_handle.write(certificate_data)
-                return file_handle
+            dump_name = "{}-{}-{}.crt".format(hive_offset, reg_section, key_hash)
+            file_handle = open_method(dump_name)
+            file_handle.write(certificate_data)
+            return file_handle
         except exceptions.InvalidAddressException:
-            vollog.debug(f"Unable to certificate file at {hive_offset:#x}")
-            return None
+            vollog.debug(f"Unable to dump certificate file at {hive_offset:#x}")
+        return None
 
 
     def _generator(self) -> Iterator[Tuple[int, Tuple[str, str, str, str]]]:
@@ -79,9 +78,10 @@ class Certificates(interfaces.plugins.PluginInterface):
                             key_hash = key_path[key_path.rindex("\\") + 1:]
 
                             if self.config['dump']:
-                                file_handle = self.dump_certificate(certificate_data, hive.hive_offset, reg_section, key_hash, self.open)
-                                if file_handle:
-                                    file_handle.close()
+                                if not isinstance(certificate_data, interfaces.renderers.BaseAbsentValue):
+                                    file_handle = self.dump_certificate(certificate_data, hive.hive_offset, reg_section, key_hash, self.open)
+                                    if file_handle:
+                                        file_handle.close()
                             
                             yield (0, (top_key, reg_section, key_hash, name))