Merge pull request #936 from volatilityfoundation/release/v2.4.1

Release  v2.4.1

```
New plugins:
   linux.sockstat
   linux.iomem
   linux.psscan
   linux.envars
   windows.drivermodule
   windows.vadwalk
Pid filtering for Windows pstree plugin
Minor fixes for Windows callbacks plugin
Minimum python version was increased to 3.7
Python-snappy dependency was replaced with ctypes to ease installation
Whole codebase was reformatted with black
Faster release cycle (targetting every 4 months)
```

.github/workflows/black.yml

@@ -6,7 +6,7 @@ jobs:
   lint:
     runs-on: ubuntu-20.04
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - uses: psf/black@stable
         with:
           options: "--check --diff --verbose"

.github/workflows/codeql.yml

@@ -17,8 +17,8 @@ on:
   pull_request:
     # The branches below must be a subset of the branches above
     branches: [ "develop" ]
-  schedule:
-    - cron: '16 8 * * 0'
+#   schedule:
+#     - cron: '16 8 * * 0'
 
 jobs:
   analyze:

README.md

@@ -107,7 +107,7 @@ The latest generated copy of the documentation can be found at: <https://volatil
 
 ## Licensing and Copyright
 
-Copyright (C) 2007-2022 Volatility Foundation
+Copyright (C) 2007-2023 Volatility Foundation
 
 All Rights Reserved
 

doc/source/simple-plugin.rst

@@ -259,7 +259,7 @@ The plugin then takes the process's ``BaseDllName`` value, and calls :py:meth:`~
 as defined by the symbols, are directly accessible and use the case-style of the symbol library it came from (in Windows,
 attributes are CamelCase), such as ``entry.BaseDllName`` in this instance.  Any attributes not defined by the symbol but added
 by Volatility extensions cannot be properties (in case they overlap with the attributes defined in the symbol libraries)
-and are therefore always methods and pretended with ``get_``, in this example ``BaseDllName.get_string()``.
+and are therefore always methods and prepended with ``get_``, in this example ``BaseDllName.get_string()``.
 
 Finally, ``FullDllName`` is populated.  These operations read from memory, and as such, the memory image may be unable to
 read the data at a particular offset.  This will cause an exception to be thrown.  In Volatility 3, exceptions are thrown

requirements-dev.txt

@@ -20,7 +20,3 @@ jsonschema>=2.3.0
 
 # This is required for memory acquisition via leechcore/pcileech.
 leechcorepyc>=2.4.0
-
-# This is required for analyzing Linux samples compressed using AVMLs native
-# compression format.  It is not required for AVML's standard LiME compression.
-python-snappy==0.6.0

requirements.txt

@@ -16,7 +16,3 @@ pycryptodome
 
 # This is required for memory acquisition via leechcore/pcileech.
 leechcorepyc>=2.4.0
-
-# This is required for analyzing Linux samples compressed using AVMLs native
-# compression format.  It is not required for AVML's standard LiME compression.
-python-snappy==0.6.0

volatility3/cli/__init__.py

@@ -443,7 +443,7 @@ class CommandLine:
             # Construct and run the plugin
             if constructed:
                 renderers[args.renderer]().render(constructed.run())
-        except (exceptions.VolatilityException) as excp:
+        except exceptions.VolatilityException as excp:
             self.process_exceptions(excp)
 
     @classmethod

volatility3/cli/text_renderer.py

@@ -241,7 +241,9 @@ class CSVRenderer(CLIRenderer):
             # Ignore the type because namedtuples don't realize they have accessible attributes
             header_list.append(f"{column.name}")
 
-        writer = csv.DictWriter(outfd, header_list, lineterminator="\n")
+        writer = csv.DictWriter(
+            outfd, header_list, lineterminator="\n", escapechar="\\"
+        )
         writer.writeheader()
 
         def visitor(node: interfaces.renderers.TreeNode, accumulator):
@@ -346,7 +348,7 @@ class PrettyTextRenderer(CLIRenderer):
 
         column_titles = [""] + [column.name for column in grid.columns]
         outfd.write(format_string.format(*column_titles))
-        for (depth, line) in final_output:
+        for depth, line in final_output:
             nums_line = max([len(line[column]) for column in line])
             for column in line:
                 line[column] = line[column] + ([""] * (nums_line - len(line[column])))

volatility3/cli/volargparse.py

@@ -31,7 +31,6 @@ class HelpfulSubparserAction(argparse._SubParsersAction):
         values: Union[str, Sequence[Any], None],
         option_string: Optional[str] = None,
     ) -> None:
-
         parser_name = ""
         arg_strings = []  # type: List[str]
         if values is not None:

volatility3/framework/automagic/construct_layers.py

@@ -36,7 +36,6 @@ class ConstructionMagic(interfaces.automagic.AutomagicInterface):
         progress_callback=None,
         optional=False,
     ) -> List[str]:
-
         # Make sure we import the layers, so they can reconstructed
         framework.import_files(sys.modules["volatility3.framework.layers"])
 

volatility3/framework/automagic/mac.py

@@ -251,7 +251,6 @@ class MacIntelStacker(interfaces.automagic.StackerLayerInterface):
             context=context,
             progress_callback=progress_callback,
         ):
-
             banner = context.layers[layer_name].read(offset, 128)
 
             idx = banner.find(b"\x00")

volatility3/framework/automagic/symbol_cache.py

@@ -161,7 +161,8 @@ class CacheManagerInterface(interfaces.configuration.VersionableInterface):
         """Returns ISF statistics based on the location
 
         Returns:
-            A tuple of base_types, types, enums, symbols, or None is location not found"""
+            A tuple of base_types, types, enums, symbols, or None is location not found
+        """
 
     def get_hash(self, location: str) -> Optional[str]:
         """Returns the hash of the JSON from within a location ISF"""
@@ -331,7 +332,7 @@ class SqliteCache(CacheManagerInterface):
                     if inner_url.scheme == "file":
                         pathname = inner_url.path.split("!")[0]
 
-                if pathname:
+                if pathname and os.path.exists(pathname):
                     timestamp = datetime.datetime.fromtimestamp(
                         os.stat(pathname).st_mtime
                     )
@@ -370,7 +371,7 @@ class SqliteCache(CacheManagerInterface):
 
                         # Get stats
                         stats_base_types = len(json_obj.get("base_types", {}))
-                        stats_types = len(json_obj.get("types", {}))
+                        stats_types = len(json_obj.get("user_types", {}))
                         stats_enums = len(json_obj.get("enums", {}))
                         stats_symbols = len(json_obj.get("symbols", {}))
 

volatility3/framework/automagic/symbol_finder.py

@@ -82,13 +82,13 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
             shortcut=False,
         )
 
-        for (sub_path, requirement) in self._requirements:
+        for sub_path, requirement in self._requirements:
             parent_path = interfaces.configuration.parent_path(sub_path)
 
             if isinstance(
                 requirement, requirements.SymbolTableRequirement
             ) and requirement.unsatisfied(context, parent_path):
-                for (tl_sub_path, tl_requirement) in self._requirements:
+                for tl_sub_path, tl_requirement in self._requirements:
                     tl_parent_path = interfaces.configuration.parent_path(tl_sub_path)
                     # Find the TranslationLayer sibling to the SymbolTableRequirement
                     if (

volatility3/framework/constants/linux/__init__.py

@@ -14,3 +14,223 @@ PAGE_SHIFT = 12
 
 # include/linux/sched.h
 PF_KTHREAD = 0x00200000  # I'm a kernel thread
+
+# Standard well-defined IP protocols.
+# ref: include/uapi/linux/in.h
+IP_PROTOCOLS = {
+    0: "IP",
+    1: "ICMP",
+    2: "IGMP",
+    4: "IPIP",
+    6: "TCP",
+    8: "EGP",
+    12: "PUP",
+    17: "UDP",
+    22: "IDP",
+    29: "TP",
+    33: "DCCP",
+    41: "IPV6",
+    46: "RSVP",
+    47: "GRE",
+    50: "ESP",
+    51: "AH",
+    92: "MTP",
+    94: "BEETPH",
+    98: "ENCAP",
+    103: "PIM",
+    108: "COMP",
+    132: "SCTP",
+    136: "UDPLITE",
+    137: "MPLS",
+    143: "ETHERNET",
+    255: "RAW",
+    262: "MPTCP",
+}
+
+# IPV6 extension headers
+# ref: include/uapi/linux/in6.h
+IPV6_PROTOCOLS = {
+    0: "HOPBYHOP_OPTS",
+    43: "ROUTING",
+    44: "FRAGMENT",
+    58: "ICMPv6",
+    59: "NO_NEXT",
+    60: "DESTINATION_OPTS",
+    135: "MOBILITY",
+}
+
+# ref: include/net/tcp_states.h
+TCP_STATES = (
+    "",
+    "ESTABLISHED",
+    "SYN_SENT",
+    "SYN_RECV",
+    "FIN_WAIT1",
+    "FIN_WAIT2",
+    "TIME_WAIT",
+    "CLOSE",
+    "CLOSE_WAIT",
+    "LAST_ACK",
+    "LISTEN",
+    "CLOSING",
+    "TCP_NEW_SYN_RECV",
+)
+
+# ref: include/linux/net.h (socket_type enum)
+SOCK_TYPES = {
+    1: "STREAM",
+    2: "DGRAM",
+    3: "RAW",
+    4: "RDM",
+    5: "SEQPACKET",
+    6: "DCCP",
+    10: "PACKET",
+}
+
+# Address families
+# ref: include/linux/socket.h
+SOCK_FAMILY = (
+    "AF_UNSPEC",
+    "AF_UNIX",
+    "AF_INET",
+    "AF_AX25",
+    "AF_IPX",
+    "AF_APPLETALK",
+    "AF_NETROM",
+    "AF_BRIDGE",
+    "AF_ATMPVC",
+    "AF_X25",
+    "AF_INET6",
+    "AF_ROSE",
+    "AF_DECnet",
+    "AF_NETBEUI",
+    "AF_SECURITY",
+    "AF_KEY",
+    "AF_NETLINK",
+    "AF_PACKET",
+    "AF_ASH",
+    "AF_ECONET",
+    "AF_ATMSVC",
+    "AF_RDS",
+    "AF_SNA",
+    "AF_IRDA",
+    "AF_PPPOX",
+    "AF_WANPIPE",
+    "AF_LLC",
+    "AF_IB",
+    "AF_MPLS",
+    "AF_CAN",
+    "AF_TIPC",
+    "AF_BLUETOOTH",
+    "AF_IUCV",
+    "AF_RXRPC",
+    "AF_ISDN",
+    "AF_PHONET",
+    "AF_IEEE802154",
+    "AF_CAIF",
+    "AF_ALG",
+    "AF_NFC",
+    "AF_VSOCK",
+    "AF_KCM",
+    "AF_QIPCRTR",
+    "AF_SMC",
+    "AF_XDP",
+)
+
+# Socket states
+# ref: include/uapi/linux/net.h
+SOCKET_STATES = ("FREE", "UNCONNECTED", "CONNECTING", "CONNECTED", "DISCONNECTING")
+
+# Netlink protocols
+# ref: include/uapi/linux/netlink.h
+NETLINK_PROTOCOLS = (
+    "NETLINK_ROUTE",
+    "NETLINK_UNUSED",
+    "NETLINK_USERSOCK",
+    "NETLINK_FIREWALL",
+    "NETLINK_SOCK_DIAG",
+    "NETLINK_NFLOG",
+    "NETLINK_XFRM",
+    "NETLINK_SELINUX",
+    "NETLINK_ISCSI",
+    "NETLINK_AUDIT",
+    "NETLINK_FIB_LOOKUP",
+    "NETLINK_CONNECTOR",
+    "NETLINK_NETFILTER",
+    "NETLINK_IP6_FW",
+    "NETLINK_DNRTMSG",
+    "NETLINK_KOBJECT_UEVENT",
+    "NETLINK_GENERIC",
+    "NETLINK_DM",
+    "NETLINK_SCSITRANSPORT",
+    "NETLINK_ECRYPTFS",
+    "NETLINK_RDMA",
+    "NETLINK_CRYPTO",
+    "NETLINK_SMC",
+)
+
+# Short list of Ethernet Protocol ID's.
+# ref: include/uapi/linux/if_ether.h
+# Used in AF_PACKET socket family
+ETH_PROTOCOLS = {
+    0x0001: "ETH_P_802_3",
+    0x0002: "ETH_P_AX25",
+    0x0003: "ETH_P_ALL",
+    0x0004: "ETH_P_802_2",
+    0x0005: "ETH_P_SNAP",
+    0x0006: "ETH_P_DDCMP",
+    0x0007: "ETH_P_WAN_PPP",
+    0x0008: "ETH_P_PPP_MP",
+    0x0009: "ETH_P_LOCALTALK",
+    0x000C: "ETH_P_CAN",
+    0x000F: "ETH_P_CANFD",
+    0x0010: "ETH_P_PPPTALK",
+    0x0011: "ETH_P_TR_802_2",
+    0x0016: "ETH_P_CONTROL",
+    0x0017: "ETH_P_IRDA",
+    0x0018: "ETH_P_ECONET",
+    0x0019: "ETH_P_HDLC",
+    0x001A: "ETH_P_ARCNET",
+    0x001B: "ETH_P_DSA",
+    0x001C: "ETH_P_TRAILER",
+    0x0060: "ETH_P_LOOP",
+    0x00F6: "ETH_P_IEEE802154",
+    0x00F7: "ETH_P_CAIF",
+    0x00F8: "ETH_P_XDSA",
+    0x00F9: "ETH_P_MAP",
+    0x0800: "ETH_P_IP",
+    0x0805: "ETH_P_X25",
+    0x0806: "ETH_P_ARP",
+    0x8035: "ETH_P_RARP",
+    0x809B: "ETH_P_ATALK",
+    0x80F3: "ETH_P_AARP",
+    0x8100: "ETH_P_8021Q",
+}
+
+# Connection and socket states
+# ref: include/net/bluetooth/bluetooth.h
+BLUETOOTH_STATES = (
+    "",
+    "CONNECTED",
+    "OPEN",
+    "BOUND",
+    "LISTEN",
+    "CONNECT",
+    "CONNECT2",
+    "CONFIG",
+    "DISCONN",
+    "CLOSED",
+)
+
+# Bluetooth protocols
+# ref: include/net/bluetooth/bluetooth.h
+BLUETOOTH_PROTOCOLS = (
+    "L2CAP",
+    "HCI",
+    "SCO",
+    "RFCOMM",
+    "BNEP",
+    "CMTP",
+    "HIDP",
+    "AVDTP",
+)

volatility3/framework/interfaces/layers.py

@@ -294,7 +294,7 @@ class DataLayerInterface(
         sections."""
         result: List[Tuple[int, int]] = []
         position = 0
-        for (start, length) in sorted(sections):
+        for start, length in sorted(sections):
             if result and start <= position:
                 initial_start, _ = result.pop()
                 result.append((initial_start, (start + length) - initial_start))
@@ -375,7 +375,6 @@ class DataLayerInterface(
     def _scan_metric(
         self, _scanner: "ScannerInterface", sections: List[Tuple[int, int]]
     ) -> Callable[[int], float]:
-
         if not sections:
             raise ValueError("Sections have no size, nothing to scan")
         last_section, last_length = sections[-1]
@@ -551,7 +550,7 @@ class TranslationLayerInterface(DataLayerInterface, metaclass=ABCMeta):
         scanner.chunk_size + scanner.overlap DataLayers by default are
         assumed to have no holes
         """
-        for (section_start, section_length) in sections:
+        for section_start, section_length in sections:
             output: List[Tuple[str, int, int]] = []
 
             # Hold the offsets of each chunk (including how much has been filled)

volatility3/framework/layers/avml.py

@@ -6,6 +6,7 @@
 
 The user of the file doesn't have to worry about the compression,
 but random access is not allowed."""
+import ctypes
 import logging
 import struct
 from typing import Tuple, List, Optional
@@ -16,13 +17,49 @@ from volatility3.framework.layers import segmented
 vollog = logging.getLogger(__name__)
 
 try:
-    import snappy
+    # TODO: Find library for windows if needed
+    try:
+        # Linux/Mac
+        lib_snappy = ctypes.cdll.LoadLibrary("libsnappy.so.1")
+    except OSError:
+        lib_snappy = None
+
+    try:
+        if not lib_snappy:
+            # Windows 64
+            lib_snappy = ctypes.cdll.LoadLibrary("snappy64")
+    except OSError:
+        lib_snappy = None
+
+    if lib_snappy:
+        # Windows 32
+        lib_snappy = ctypes.cdll.LoadLibrary("snappy32")
+
+    __snappy_uncompress = lib_snappy.snappy_uncompress
+    __snappy_uncompressed_length = lib_snappy.snappy_uncompressed_length
 
     HAS_SNAPPY = True
-except ImportError:
+except (AttributeError, OSError):
     HAS_SNAPPY = False
 
 
+class SnappyException(exceptions.VolatilityException):
+    pass
+
+
+def uncompress(s):
+    """Uncompress a snappy compressed string."""
+    ulen = ctypes.c_int(0)
+    cresult = __snappy_uncompressed_length(s, len(s), ctypes.byref(ulen))
+    if cresult != 0:
+        raise SnappyException(f"Error in snappy_uncompressed_length: {cresult}")
+    ubuf = ctypes.create_string_buffer(ulen.value)
+    cresult = __snappy_uncompress(s, len(s), ubuf, ctypes.byref(ulen))
+    if cresult != 0:
+        raise SnappyException(f"Error in snappy_uncompress: {cresult}")
+    return ubuf.raw
+
+
 class AVMLLayer(segmented.NonLinearlySegmentedLayer):
     """A Lime format TranslationLayer.
 
@@ -42,10 +79,11 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer):
             layer.read(layer.minimum_address, struct.calcsize(header_structure)),
         )
         if magic not in [0x4C4D5641] or version != 2:
-            raise exceptions.LayerException("File not completely in AVML format")
+            raise exceptions.LayerException("File not in AVML format")
         if not HAS_SNAPPY:
             vollog.warning(
-                "AVML file detected, but snappy python library not installed"
+                "AVML file detected, but snappy library could not be found\n"
+                "Please install the snappy from your distribution or https://google.github.io/snappy/."
             )
             raise exceptions.LayerException(
                 "AVML format dependencies not satisfied (snappy)"
@@ -73,7 +111,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer):
             )
             segments, consumed = self._read_snappy_frames(chunk_data, end - start)
             # The returned segments are accurate the chunk_data that was passed in, but needs shifting
-            for (thing, mapped_offset, size, mapped_size, compressed) in segments:
+            for thing, mapped_offset, size, mapped_size, compressed in segments:
                 self._segments.append(
                     (
                         thing + start,
@@ -131,7 +169,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer):
                     ]
                     if frame_type == 0x00:
                         # Compressed data
-                        frame_data = snappy.decompress(frame_data)
+                        frame_data = uncompress(frame_data)
                     # TODO: Verify CRC
                     segments.append(
                         (
@@ -156,7 +194,7 @@ class AVMLLayer(segmented.NonLinearlySegmentedLayer):
     ) -> bytes:
         start_offset, _, _, _ = self._find_segment(offset)
         if self._compressed[mapped_offset]:
-            decoded_data = snappy.decompress(data)
+            decoded_data = uncompress(data)
         else:
             decoded_data = data
         decoded_data = decoded_data[offset - start_offset :]

volatility3/framework/layers/crash.py

@@ -39,7 +39,6 @@ class WindowsCrashDump32Layer(segmented.SegmentedLayer):
     def __init__(
         self, context: interfaces.context.ContextInterface, config_path: str, name: str
     ) -> None:
-
         # Construct these so we can use self.config
         self._context = context
         self._config_path = config_path

volatility3/framework/layers/elf.py

@@ -119,4 +119,8 @@ class Elf64Stacker(interfaces.automagic.StackerLayerInterface):
             interfaces.configuration.path_join(new_name, "base_layer")
         ] = layer_name
 
-        return Elf64Layer(context, new_name, new_name)
+        try:
+            return Elf64Layer(context, new_name, new_name)
+        except ElfFormatException as excp:
+            vollog.log(constants.LOGLEVEL_VVVV, f"Exception: {excp}")
+            return None

volatility3/framework/layers/intel.py

@@ -56,6 +56,11 @@ class Intel(linear.LinearlyMappedLayer):
         )
         self._entry_size = struct.calcsize(self._entry_format)
         self._entry_number = self.page_size // self._entry_size
+        self._canonical_prefix = self._mask(
+            (1 << self._bits_per_register) - 1,
+            self._bits_per_register,
+            self._maxvirtaddr,
+        )
 
         # These can vary depending on the type of space
         self._index_shift = int(
@@ -106,6 +111,23 @@ class Intel(linear.LinearlyMappedLayer):
         """Returns whether a particular page is valid based on its entry."""
         return bool(entry & 1)
 
+    def canonicalize(self, addr: int) -> int:
+        """Canonicalizes an address by performing an appropiate sign extension on the higher addresses"""
+        if self._bits_per_register <= self._maxvirtaddr:
+            return addr & self.address_mask
+        elif addr < (1 << self._maxvirtaddr - 1):
+            return addr
+        return self._mask(addr, self._maxvirtaddr, 0) + self._canonical_prefix
+
+    def decanonicalize(self, addr: int) -> int:
+        """Removes canonicalization to ensure an adress fits within the correct range if it has been canonicalized
+
+        This will produce an address outside the range if the canonicalization is incorrect
+        """
+        if addr < (1 << self._maxvirtaddr - 1):
+            return addr
+        return addr ^ self._canonical_prefix
+
     def _translate(self, offset: int) -> Tuple[int, int, str]:
         """Translates a specific offset based on paging tables.
 
@@ -151,7 +173,7 @@ class Intel(linear.LinearlyMappedLayer):
             )
 
         # Run through the offset in various chunks
-        for (name, size, large_page) in self._structure:
+        for name, size, large_page in self._structure:
             # Check we're valid
             if not self._page_is_valid(entry):
                 raise exceptions.PagedInvalidAddressException(

volatility3/framework/layers/leechcore.py

@@ -91,7 +91,6 @@ if HAS_LEECHCORE:
             chunk_size = size
             output = []
             for entry in self.handle.memmap:
-
                 if (
                     entry["base"] + entry["size"] <= chunk_start
                     or entry["base"] >= chunk_start + chunk_size

volatility3/framework/layers/linear.py

@@ -42,7 +42,7 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
         length size."""
         current_offset = offset
         output: List[bytes] = []
-        for (offset, _, mapped_offset, mapped_length, layer) in self.mapping(
+        for offset, _, mapped_offset, mapped_length, layer in self.mapping(
             offset, length, ignore_errors=pad
         ):
             if not pad and offset > current_offset:
@@ -71,7 +71,7 @@ class LinearlyMappedLayer(interfaces.layers.TranslationLayerInterface):
         underlying mapping."""
         current_offset = offset
         length = len(value)
-        for (offset, _, mapped_offset, length, layer) in self.mapping(offset, length):
+        for offset, _, mapped_offset, length, layer in self.mapping(offset, length):
             if offset > current_offset:
                 raise exceptions.InvalidAddressException(
                     self.name,

volatility3/framework/layers/qemu.py

@@ -117,13 +117,15 @@ class QemuSuspendLayer(segmented.NonLinearlySegmentedLayer):
         chunk_size = 4096
         data = b""
         for i in range(
-            base_layer.maximum_address, base_layer.minimum_address, -chunk_size
+            base_layer.maximum_address + 1, base_layer.minimum_address, -chunk_size
         ):
-            if i != base_layer.maximum_address:
+            # Since we're going backwards, we need to include one extra byte so the tail doesn't get chopped off
+            if i != base_layer.maximum_address + 1:
                 data = (base_layer.read(i, chunk_size) + data).rstrip(b"\x00")
                 if b"\x00" in data:
                     last_null_byte = data.rfind(b"\x00")
                     start_of_json = data.find(b"{", last_null_byte)
+
                     if start_of_json >= 0:
                         data = data[start_of_json:]
                         return json.loads(data)

volatility3/framework/layers/registry.py

@@ -269,7 +269,6 @@ class RegistryHive(linear.LinearlyMappedLayer):
     def mapping(
         self, offset: int, length: int, ignore_errors: bool = False
     ) -> Iterable[Tuple[int, int, int, int, str]]:
-
         if length < 0:
             raise ValueError("Mapping length of RegistryHive must be positive or zero")
 

volatility3/framework/objects/__init__.py

@@ -442,7 +442,7 @@ class Pointer(Integer):
     def is_readable(self, layer_name: Optional[str] = None) -> bool:
         """Determines whether the address of this pointer can be read from
         memory."""
-        layer_name = layer_name or self.vol.layer_name
+        layer_name = layer_name or self.vol.native_layer_name
         return self._context.layers[layer_name].is_valid(self, self.vol.subtype.size)
 
     def __getattr__(self, attr: str) -> Any:

volatility3/framework/plugins/linux/check_afinfo.py

@@ -68,7 +68,6 @@ class Check_afinfo(plugins.PluginInterface):
             yield var_name, "show", var.seq_show
 
     def _generator(self):
-
         vmlinux = self.context.modules[self.config["kernel"]]
 
         op_members = vmlinux.get_type("file_operations").members
@@ -86,7 +85,7 @@ class Check_afinfo(plugins.PluginInterface):
         )
         protocols = [tcp, udp]
 
-        for (struct_type, global_vars) in protocols:
+        for struct_type, global_vars in protocols:
             for global_var_name in global_vars:
                 # this will lookup fail for the IPv6 protocols on kernels without IPv6 support
                 try:
@@ -104,7 +103,6 @@ class Check_afinfo(plugins.PluginInterface):
                     yield 0, (name, member, format_hints.Hex(address))
 
     def run(self):
-
         return renderers.TreeGrid(
             [
                 ("Symbol Name", str),

volatility3/framework/plugins/linux/check_creds.py

@@ -46,7 +46,6 @@ class Check_creds(interfaces.plugins.PluginInterface):
         tasks = pslist.PsList.list_tasks(self.context, vmlinux.name)
 
         for task in tasks:
-
             cred_addr = task.cred.dereference().vol.offset
 
             if cred_addr not in creds:
@@ -54,7 +53,7 @@ class Check_creds(interfaces.plugins.PluginInterface):
 
             creds[cred_addr].append(task.pid)
 
-        for (_, pids) in creds.items():
+        for _, pids in creds.items():
             if len(pids) > 1:
                 pid_str = ""
                 for pid in pids:

volatility3/framework/plugins/linux/check_modules.py

@@ -37,7 +37,6 @@ class Check_modules(plugins.PluginInterface):
     def get_kset_modules(
         cls, context: interfaces.context.ContextInterface, vmlinux_name: str
     ):
-
         vmlinux = context.modules[vmlinux_name]
 
         try:
@@ -57,7 +56,6 @@ class Check_modules(plugins.PluginInterface):
         for kobj in module_kset.list.to_list(
             vmlinux.symbol_table_name + constants.BANG + "kobject", "entry"
         ):
-
             mod_kobj = vmlinux.object(
                 object_type="module_kobject",
                 offset=kobj.vol.offset - kobj_off,

volatility3/framework/plugins/linux/check_syscall.py

@@ -110,7 +110,7 @@ class Check_syscall(plugins.PluginInterface):
         vmlinux = self.context.modules[self.config["kernel"]]
         data = self.context.layers.read(vmlinux.layer_name, func_addr, 6)
 
-        for (address, size, mnemonic, op_str) in md.disasm_lite(data, func_addr):
+        for address, size, mnemonic, op_str in md.disasm_lite(data, func_addr):
             if mnemonic == "CMP":
                 table_size = int(op_str.split(",")[1].strip()) & 0xFFFF
                 break
@@ -161,7 +161,7 @@ class Check_syscall(plugins.PluginInterface):
             ia32_info = self._get_table_info(vmlinux, "ia32_sys_call_table", ptr_sz)
             tables.append(("32bit", ia32_info))
 
-        for (table_name, (tableaddr, tblsz)) in tables:
+        for table_name, (tableaddr, tblsz) in tables:
             table = vmlinux.object(
                 object_type="array",
                 subtype=vmlinux.get_type("pointer"),
@@ -169,7 +169,7 @@ class Check_syscall(plugins.PluginInterface):
                 count=tblsz,
             )
 
-            for (i, call_addr) in enumerate(table):
+            for i, call_addr in enumerate(table):
                 if not call_addr:
                     continue
 
@@ -196,7 +196,6 @@ class Check_syscall(plugins.PluginInterface):
                 )
 
     def run(self):
-
         return renderers.TreeGrid(
             [
                 ("Table Address", format_hints.Hex),

volatility3/framework/plugins/linux/envars.py

@@ -0,0 +1,121 @@
+# This file is Copyright 2022 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+
+import logging
+
+from volatility3.framework import exceptions, renderers
+from volatility3.framework.configuration import requirements
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.objects import utility
+from volatility3.plugins.linux import pslist
+
+vollog = logging.getLogger(__name__)
+
+
+class Envars(plugins.PluginInterface):
+    """Lists processes with their environment variables"""
+
+    _required_framework_version = (2, 0, 0)
+
+    @classmethod
+    def get_requirements(cls):
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(
+                name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
+            ),
+            requirements.ListRequirement(
+                name="pid",
+                description="Filter on specific process IDs",
+                element_type=int,
+                optional=True,
+            ),
+        ]
+
+    def _generator(self, tasks):
+        """Generates a listing of processes along with environment variables"""
+
+        # walk the process list and return the envars
+        for task in tasks:
+            pid = task.pid
+
+            # get process name as string
+            name = utility.array_to_string(task.comm)
+
+            # try and get task parent
+            try:
+                ppid = task.parent.pid
+            except exceptions.InvalidAddressException:
+                vollog.debug(
+                    f"Unable to read parent pid for task {pid} {name}, setting ppid to 0."
+                )
+                ppid = 0
+
+            # kernel threads never have an mm as they do not have userland mappings
+            try:
+                mm = task.mm
+            except exceptions.InvalidAddressException:
+                # no mm so cannot get envars
+                vollog.debug(
+                    f"Unable to access mm for task {pid} {name} it is likely a kernel thread, will not extract any envars."
+                )
+                mm = None
+                continue
+
+            # if mm exists attempt to get envars
+            if mm:
+                # get process layer to read envars from
+                proc_layer_name = task.add_process_layer()
+                if proc_layer_name is None:
+                    vollog.debug(
+                        f"Unable to construct process layer for task {pid} {name}, will not extract any envars."
+                    )
+                    continue
+                proc_layer = self.context.layers[proc_layer_name]
+
+                # get the size of the envars with sanity checking
+                envars_size = task.mm.env_end - task.mm.env_start
+                if not (0 < envars_size <= 8192):
+                    vollog.debug(
+                        f"Task {pid} {name} appears to have envars of size {envars_size} bytes which fails the sanity checking, will not extract any envars."
+                    )
+                    continue
+
+                # attempt to read all envars data
+                try:
+                    envar_data = proc_layer.read(task.mm.env_start, envars_size)
+                except exceptions.InvalidAddressException:
+                    vollog.debug(
+                        f"Unable to read full envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)} for {envars_size} bytes, will not extract any envars."
+                    )
+                    continue
+
+                # parse envar data, envars are null terminated, keys and values are separated by '='
+                envar_data = envar_data.rstrip(b"\x00")
+                for envar_pair in envar_data.split(b"\x00"):
+                    try:
+                        key, value = envar_pair.decode().split("=", 1)
+                    except ValueError:
+                        vollog.debug(
+                            f"Unable to extract envars for {pid} {name} starting at virtual offset {hex(task.mm.env_start)}, they don't appear to be '=' separated"
+                        )
+                        continue
+                    yield (0, (pid, ppid, name, key, value))
+
+    def run(self):
+        filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
+
+        return renderers.TreeGrid(
+            [("PID", int), ("PPID", int), ("COMM", str), ("KEY", str), ("VALUE", str)],
+            self._generator(
+                pslist.PsList.list_tasks(
+                    self.context, self.config["kernel"], filter_func=filter_func
+                )
+            ),
+        )

volatility3/framework/plugins/linux/iomem.py

@@ -0,0 +1,161 @@
+# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+import logging
+from typing import List
+
+from volatility3.framework import renderers, interfaces, exceptions
+from volatility3.framework.configuration import requirements
+from volatility3.framework.objects import utility
+from volatility3.framework.renderers import format_hints
+
+vollog = logging.getLogger(__name__)
+
+
+class IOMem(interfaces.plugins.PluginInterface):
+    """Generates an output similar to /proc/iomem on a running system."""
+
+    _required_framework_version = (2, 0, 0)
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            )
+        ]
+
+    @classmethod
+    def parse_resource(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+        resource_offset: int,
+        seen: set = set(),
+        depth: int = 0,
+    ):
+        """Recursively parse from a root resource to find details about all related resources.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+            resource_offset: The offset to the resouce to be parsed
+            seen: The set of resource offsets that have already been parsed
+            depth: How deep into the resource structure we are
+
+        Yields:
+            Each row of output
+        """
+        vmlinux = context.modules[vmlinux_module_name]
+
+        # create the resource object with protection against memory smear
+        try:
+            resource = vmlinux.object("resource", resource_offset)
+        except exceptions.InvalidAddressException:
+            vollog.warning(
+                f"Unable to create resource object at {resource_offset:#x}. This resource, "
+                "its sibling, and any of it's childern and will be missing from the output."
+            )
+            return None
+
+        # get name with protection against smear as following a pointer
+        try:
+            name = utility.pointer_to_string(resource.name, 128)
+        except exceptions.InvalidAddressException:
+            vollog.warning(
+                "Unable to follow pointer to name for resource object at {resource_offset:#x}, "
+                "replaced with UnreadableValue"
+            )
+            name = renderers.UnreadableValue()
+
+        # mark this resource as seen in the seen set. Normally this should not be needed but will protect
+        # against possible infinite loops. Warn the user if an infinite loop would have happened.
+        if resource_offset in seen:
+            vollog.warning(
+                f"The resource object at {resource_offset:#x} '{name}' has already been processed, "
+                "this should not normally occur. No further results from related resources will be "
+                "displayed to protect against infinite loops."
+            )
+            return None
+        else:
+            seen.add(resource_offset)
+
+        # yield information on this resource
+        yield depth, (name, resource.start, resource.end)
+
+        # process child resource if this exists
+        if resource.child != 0:
+            yield from cls.parse_resource(
+                context,
+                vmlinux_module_name,
+                resource.child,
+                seen,
+                depth + 1,
+            )
+
+        # process sibling resource if this exists
+        if resource.sibling != 0:
+            yield from cls.parse_resource(
+                context,
+                vmlinux_module_name,
+                resource.sibling,
+                seen,
+                depth,
+            )
+
+    def _generator(self):
+        """Generates an output similar to /proc/iomem on a running system
+
+        Args:
+            None
+
+        Yields:
+            Each row of output using the parse_resource function
+        """
+
+        # get the kernel module from the current context
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+
+        # get the address for the iomem_resource
+        try:
+            iomem_root_offset = vmlinux.get_absolute_symbol_address("iomem_resource")
+        except exceptions.SymbolError:
+            iomem_root_offset = None
+
+        # only continue if iomem_root address was located
+        if iomem_root_offset is not None:
+            # recursively parse the resources starting from the root resource at 'iomem_resource'
+            for depth, (name, start, end) in self.parse_resource(
+                self.context, vmlinux_module_name, iomem_root_offset
+            ):
+                # use format_hints to format start and end addresses for the renderers
+                yield depth, (name, format_hints.Hex(start), format_hints.Hex(end))
+
+    def run(self):
+        # get the kernel module from the current context
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+
+        # check that the iomem_resource symbol exists
+        # normally exported in /kernel/resource.c
+        if not vmlinux.has_symbol("iomem_resource"):
+            raise TypeError(
+                "This plugin requires the iomem_resource symbol. This symbol is not present in the supplied symbol table. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt."
+            )
+
+        # error if type 'resource' is not found
+        if not vmlinux.has_type("resource"):
+            raise TypeError(
+                "This plugin requires the resource type. This type is not present in the supplied symbol table. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt."
+            )
+
+        columns = [
+            ("Name", str),
+            ("Start", format_hints.Hex),
+            ("End", format_hints.Hex),
+        ]
+        return renderers.TreeGrid(columns, self._generator())

volatility3/framework/plugins/linux/lsmod.py

@@ -60,7 +60,6 @@ class Lsmod(plugins.PluginInterface):
     def _generator(self):
         try:
             for module in self.list_modules(self.context, self.config["kernel"]):
-
                 mod_size = module.get_init_size() + module.get_core_size()
 
                 mod_name = utility.array_to_string(module.name)

volatility3/framework/plugins/linux/lsof.py

@@ -4,7 +4,7 @@
 """A module containing a collection of plugins that produce data typically
 found in Linux's /proc file system."""
 import logging
-from typing import List
+from typing import List, Callable
 
 from volatility3.framework import renderers, interfaces, constants
 from volatility3.framework.configuration import requirements
@@ -21,6 +21,8 @@ class Lsof(plugins.PluginInterface):
 
     _required_framework_version = (2, 0, 0)
 
+    _version = (1, 1, 0)
+
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
         return [
@@ -43,34 +45,45 @@ class Lsof(plugins.PluginInterface):
             ),
         ]
 
-    def _generator(self, tasks):
-        symbol_table = None
-        for task in tasks:
-            if symbol_table is None:
+    @classmethod
+    def list_fds(
+        cls,
+        context: interfaces.context.ContextInterface,
+        symbol_table: str,
+        filter_func: Callable[[int], bool] = lambda _: False,
+    ):
+        linuxutils_symbol_table = None  # type: ignore
+        for task in pslist.PsList.list_tasks(context, symbol_table, filter_func):
+            if linuxutils_symbol_table is None:
                 if constants.BANG not in task.vol.type_name:
                     raise ValueError("Task is not part of a symbol table")
-                symbol_table = task.vol.type_name.split(constants.BANG)[0]
+                linuxutils_symbol_table = task.vol.type_name.split(constants.BANG)[0]
 
-            name = utility.array_to_string(task.comm)
+            task_comm = utility.array_to_string(task.comm)
             pid = int(task.pid)
 
-            for (
-                fd_num,
-                _,
-                full_path,
-            ) in linux.LinuxUtilities.files_descriptors_for_process(
-                self.context, symbol_table, task
-            ):
-                yield (0, (pid, name, fd_num, full_path))
+            fd_generator = linux.LinuxUtilities.files_descriptors_for_process(
+                context, linuxutils_symbol_table, task
+            )
+
+            for fd_fields in fd_generator:
+                yield pid, task_comm, task, fd_fields
+
+    def _generator(self, pids, symbol_table):
+        filter_func = pslist.PsList.create_pid_filter(pids)
+        fds_generator = self.list_fds(
+            self.context, symbol_table, filter_func=filter_func
+        )
+
+        for pid, task_comm, _task, fd_fields in fds_generator:
+            fd_num, _filp, full_path = fd_fields
+
+            fields = (pid, task_comm, fd_num, full_path)
+            yield (0, fields)
 
     def run(self):
-        filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
+        pids = self.config.get("pid", None)
+        symbol_table = self.config["kernel"]
 
-        return renderers.TreeGrid(
-            [("PID", int), ("Process", str), ("FD", int), ("Path", str)],
-            self._generator(
-                pslist.PsList.list_tasks(
-                    self.context, self.config["kernel"], filter_func=filter_func
-                )
-            ),
-        )
+        tree_grid_args = [("PID", int), ("Process", str), ("FD", int), ("Path", str)]
+        return renderers.TreeGrid(tree_grid_args, self._generator(pids, symbol_table))

volatility3/framework/plugins/linux/mountinfo.py

@@ -203,7 +203,6 @@ class MountInfo(plugins.PluginInterface):
         mount_format: bool,
         per_namespace: bool,
     ) -> Iterable[Tuple[int, Tuple]]:
-
         for task, mnt, mnt_ns_id in self._get_tasks_mountpoints(tasks, per_namespace):
             if mnt_ns_ids and mnt_ns_id not in mnt_ns_ids:
                 continue

volatility3/framework/plugins/linux/psscan.py

@@ -0,0 +1,179 @@
+# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+import logging
+from typing import Iterable, List, Tuple
+import struct
+from enum import Enum
+
+from volatility3.framework import renderers, interfaces, symbols, constants, exceptions
+from volatility3.framework.configuration import requirements
+from volatility3.framework.objects import utility
+from volatility3.framework.layers import scanners
+from volatility3.framework.renderers import format_hints
+
+vollog = logging.getLogger(__name__)
+
+
+class DescExitStateEnum(Enum):
+    """Enum for linux task exit_state as defined in include/linux/sched.h"""
+
+    TASK_RUNNING = 0x00000000
+    EXIT_DEAD = 0x00000010
+    EXIT_ZOMBIE = 0x00000020
+    EXIT_TRACE = EXIT_ZOMBIE | EXIT_DEAD
+
+
+class PsScan(interfaces.plugins.PluginInterface):
+    """Scans for processes present in a particular linux image."""
+
+    _required_framework_version = (2, 0, 0)
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+        ]
+
+    def _get_task_fields(
+        self, task: interfaces.objects.ObjectInterface
+    ) -> Tuple[int, int, int, str, str]:
+        """Extract the fields needed for the final output
+
+        Args:
+            task: A task object from where to get the fields.
+        Returns:
+            A tuple with the fields to show in the plugin output.
+        """
+        pid = task.tgid
+        tid = task.pid
+        ppid = task.parent.tgid if task.parent else 0
+        name = utility.array_to_string(task.comm)
+        exit_state = DescExitStateEnum(task.exit_state).name
+
+        task_fields = (
+            format_hints.Hex(task.vol.offset),
+            pid,
+            tid,
+            ppid,
+            name,
+            exit_state,
+        )
+        return task_fields
+
+    def _generator(self):
+        """Generates the tasks found from scanning."""
+
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+
+        for task in self.scan_tasks(
+            self.context, vmlinux_module_name, vmlinux.layer_name
+        ):
+            row = self._get_task_fields(task)
+            yield (0, row)
+
+    @classmethod
+    def scan_tasks(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+        kernel_layer_name: str,
+    ) -> Iterable[interfaces.objects.ObjectInterface]:
+        """Scans for tasks in the memory layer.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+            kernel_layer_name: The name for the kernel layer
+        Yields:
+            Task objects
+        """
+        vmlinux = context.modules[vmlinux_module_name]
+
+        # check if this image is 32bit or 64bit
+        is_32bit = not symbols.symbol_table_is_64bit(context, vmlinux.symbol_table_name)
+        if is_32bit:
+            pack_format = "I"
+        else:
+            pack_format = "Q"
+        # get task_struct to find the offset to the sched_class pointer
+        sched_class_offset = vmlinux.get_type("task_struct").members["sched_class"][0]
+        kernel_layer = context.layers[kernel_layer_name]
+
+        needles = []
+        for symbol in vmlinux.symbols:
+            # find all sched_class names by searching by if they include '_sched_class', e.g. 'fair_sched_class'
+            if "_sched_class" in symbol:
+                # use canonicalize to set the appropriate sign extension for the addr
+                addr = kernel_layer.canonicalize(
+                    vmlinux.get_symbol(symbol).address + vmlinux.offset
+                )
+                packed_addr = struct.pack(pack_format, addr)
+
+                # debug message to show needles being searched for and symbol names
+                vollog.debug(
+                    f"Found a sched_class named {symbol} at offset {hex(addr)}. Will scan for these bytes: {packed_addr.hex()}"
+                )
+
+                # append to needles list the packed hex for searching
+                needles.append(packed_addr)
+        # find the memory layer to scan
+        if len(kernel_layer.dependencies) > 1:
+            vollog.warning(
+                f"Kernel layer depends on multiple layers however only {kernel_layer.dependencies[0]} will be scanned by this plugin."
+            )
+        elif len(kernel_layer.dependencies) == 0:
+            vollog.error(
+                f"Kernel layer has no dependencies, meaning there is no memory layer for this plugin to scan."
+            )
+            raise exceptions.LayerException(
+                kernel_layer_name, f"Layer {kernel_layer_name} has no dependencies"
+            )
+        memory_layer_name = kernel_layer.dependencies[0]
+        memory_layer = context.layers[kernel_layer.dependencies[0]]
+
+        # scan the memory_layer for these needles
+        for address, _ in memory_layer.scan(
+            context, scanners.MultiStringScanner(needles)
+        ):
+            # create task in the memory_layer
+            ptask = context.object(
+                vmlinux.symbol_table_name + constants.BANG + "task_struct",
+                offset=address - sched_class_offset,
+                layer_name=memory_layer_name,
+                native_layer_name=kernel_layer_name,
+            )
+
+            # sanity check exit_state
+            try:
+                # attempt tp parse the exist_state using the enum
+                DescExitStateEnum(ptask.exit_state)
+            except ValueError:
+                vollog.debug(
+                    f"Skipping task_struct at {hex(ptask.vol.offset)} as exit_state {ptask.exit_state} is likely not valid"
+                )
+                continue
+            # sanity check pid
+            if not (0 < ptask.pid < 65535):
+                vollog.debug(
+                    f"Skipping task_struct at {hex(ptask.vol.offset)} as pid {ptask.pid} is likely not valid"
+                )
+                continue
+            yield ptask
+
+    def run(self):
+        columns = [
+            ("OFFSET (P)", format_hints.Hex),
+            ("PID", int),
+            ("TID", int),
+            ("PPID", int),
+            ("COMM", str),
+            ("EXIT_STATE", str),
+        ]
+        return renderers.TreeGrid(columns, self._generator())

volatility3/framework/plugins/linux/sockstat.py

@@ -0,0 +1,624 @@
+# This file is Copyright 2021 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+
+import logging
+from typing import Callable, Tuple, List, Dict
+
+from volatility3.framework import interfaces, exceptions, constants, objects
+from volatility3.framework.renderers import TreeGrid, NotAvailableValue, format_hints
+from volatility3.framework.configuration import requirements
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.objects import utility
+from volatility3.framework.symbols import linux
+from volatility3.plugins.linux import lsof
+
+
+vollog = logging.getLogger(__name__)
+
+
+class SockHandlers(interfaces.configuration.VersionableInterface):
+    """Handles several socket families extracting the sockets information."""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 0)
+
+    def __init__(self, vmlinux, task):
+        self._vmlinux = vmlinux
+        self._task = task
+
+        netns_id = task.nsproxy.net_ns.get_inode()
+        self._netdevices = self._build_network_devices_map(netns_id)
+
+        self._sock_family_handlers = {
+            "AF_UNIX": self._unix_sock,
+            "AF_INET": self._inet_sock,
+            "AF_INET6": self._inet_sock,
+            "AF_NETLINK": self._netlink_sock,
+            "AF_VSOCK": self._vsock_sock,
+            "AF_PACKET": self._packet_sock,
+            "AF_XDP": self._xdp_sock,
+            "AF_BLUETOOTH": self._bluetooth_sock,
+        }
+
+    def _build_network_devices_map(self, netns_id: int) -> Dict:
+        """Given a namespace ID it returns a dictionary mapping each network
+        interface index (ifindex) to its network interface name:
+
+        Args:
+            netns_id: The network namespace ID
+
+        Returns:
+            netdevices_map: Mapping network interface index (ifindex) to network
+                            interface name
+        """
+        netdevices_map = {}
+        nethead = self._vmlinux.object_from_symbol(symbol_name="net_namespace_list")
+        net_symname = self._vmlinux.symbol_table_name + constants.BANG + "net"
+        for net in nethead.to_list(net_symname, "list"):
+            net_device_symname = (
+                self._vmlinux.symbol_table_name + constants.BANG + "net_device"
+            )
+            for net_dev in net.dev_base_head.to_list(net_device_symname, "dev_list"):
+                if net.get_inode() != netns_id:
+                    continue
+                dev_name = utility.array_to_string(net_dev.name)
+                netdevices_map[net_dev.ifindex] = dev_name
+        return netdevices_map
+
+    def process_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str], Dict]:
+        """Takes a kernel generic `sock` object and processes it with its respective socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns a tuple with:
+            sock: The respective kernel's *_sock object for that socket family
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+            socket_filter: A dictionary with information about the socket filter
+        """
+        family = sock.get_family()
+        socket_filter = {}
+        sock_handler = self._sock_family_handlers.get(family)
+        if sock_handler:
+            try:
+                unix_sock, sock_stat = sock_handler(sock)
+                self._update_socket_filters_info(sock, socket_filter)
+
+                return unix_sock, sock_stat, socket_filter
+            except exceptions.SymbolError as e:
+                # Cannot finds the *_sock type in the symbols
+                vollog.log(
+                    constants.LOGLEVEL_V,
+                    "Error processing socket family '%s': %s",
+                    family,
+                    e,
+                )
+        else:
+            vollog.log(constants.LOGLEVEL_V, "Unsupported family '%s'", family)
+
+        # Even if the sock family is not supported, or the required types
+        # are not present in the symbols, we can still show some general
+        # information about the socket that may be helpful.
+        src_addr = src_port = dst_addr = dst_port = None
+        state = sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+
+        return sock, sock_stat, socket_filter
+
+    def _update_socket_filters_info(
+        self, sock: objects.Pointer, socket_filter: dict
+    ) -> None:
+        """Get information from the socket and reuseport filters
+
+        Args:
+            sock: The kernel sock (sk) struct
+            socket_filter: A dictionary with information about the socket filter
+        """
+        if sock.has_member("sk_filter") and sock.sk_filter:
+            sock_filter = sock.sk_filter
+            socket_filter["filter_type"] = "socket_filter"
+            self._extract_socket_filter_info(sock_filter, socket_filter)
+
+        if sock.has_member("sk_reuseport_cb") and sock.sk_reuseport_cb:
+            sock_reuseport_cb = sock.sk_reuseport_cb
+            socket_filter["filter_type"] = "reuseport_filter"
+            self._extract_socket_filter_info(sock_reuseport_cb, socket_filter)
+
+    def _extract_socket_filter_info(
+        self, sock_filter: objects.Pointer, socket_filter: dict
+    ) -> None:
+        """Get specific information for each type of filter
+
+        Args:
+            socket_filter: A dictionary with information about the socket filter
+        """
+        socket_filter["bpf_filter_type"] = "cBPF"
+
+        if not sock_filter.has_member("prog") or not sock_filter.prog:
+            return
+
+        bpfprog = sock_filter.prog
+        if bpfprog.type == 0:
+            # BPF_PROG_TYPE_UNSPEC = 0
+            return
+
+        socket_filter["bpf_filter_type"] = "eBPF"
+        if not bpfprog.has_member("aux") or not bpfprog.aux:
+            return
+        bpfprog_aux = bpfprog.aux
+        if bpfprog_aux.has_member("id"):
+            # `id` member was added to `bpf_prog_aux` in kernels 4.13
+            socket_filter["bpf_filter_id"] = str(bpfprog_aux.id)
+        if bpfprog_aux.has_member("name"):
+            # `name` was added to `bpf_prog_aux` in kernels 4.15
+            bpfprog_name = utility.array_to_string(bpfprog_aux.name)
+            if bpfprog_name:
+                socket_filter["bpf_filter_name"] = bpfprog_name
+
+    def _unix_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_UNIX socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            unix_sock: The kernel's `unix_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        unix_sock = sock.cast("unix_sock")
+        state = unix_sock.get_state()
+        src_addr = unix_sock.get_name()
+        src_port = unix_sock.get_inode()
+
+        if unix_sock.peer:
+            peer = unix_sock.peer.dereference().cast("unix_sock")
+            dst_addr = peer.get_name()
+            dst_port = peer.get_inode()
+        else:
+            dst_addr = dst_port = None
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return unix_sock, sock_stat
+
+    def _inet_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_INET/6 socket families
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            inet_sock: The kernel's `inet_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        inet_sock = sock.cast("inet_sock")
+        src_addr = inet_sock.get_src_addr()
+        src_port = inet_sock.get_src_port()
+        dst_addr = inet_sock.get_dst_addr()
+        dst_port = inet_sock.get_dst_port()
+        state = inet_sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return inet_sock, sock_stat
+
+    def _netlink_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_NETLINK socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            netlink_sock: The kernel's `netlink_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        netlink_sock = sock.cast("netlink_sock")
+
+        src_addr = None
+        if netlink_sock.groups:
+            groups_bitmap = netlink_sock.groups.dereference()
+            src_addr = f"groups:0x{groups_bitmap:08x}"
+        src_port = netlink_sock.portid
+
+        dst_addr = f"group:0x{netlink_sock.dst_group:08x}"
+        module = netlink_sock.module
+        if module and module.name:
+            module_name_str = utility.array_to_string(module.name)
+            dst_addr = f"{dst_addr},lkm:{module_name_str}"
+        dst_port = netlink_sock.dst_portid
+
+        state = netlink_sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return netlink_sock, sock_stat
+
+    def _vsock_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_VSOCK socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            vsock_sock: The kernel `vsock_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        vsock_sock = sock.cast("vsock_sock")
+        src_addr = vsock_sock.local_addr.svm_cid
+        src_port = vsock_sock.local_addr.svm_port
+        dst_addr = vsock_sock.remote_addr.svm_cid
+        dst_port = vsock_sock.remote_addr.svm_port
+        state = vsock_sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return vsock_sock, sock_stat
+
+    def _packet_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_PACKET socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            packet_sock: The kernel's `packet_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        packet_sock = sock.cast("packet_sock")
+        ifindex = packet_sock.ifindex
+        dev_name = self._netdevices.get(ifindex) if ifindex > 0 else "ANY"
+
+        src_addr = dev_name
+        src_port = dst_addr = dst_port = None
+        state = packet_sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return packet_sock, sock_stat
+
+    def _xdp_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_XDP socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            xdp_sock: The kernel's `xdp_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        xdp_sock = sock.cast("xdp_sock")
+        device = xdp_sock.dev
+        if not device:
+            return
+
+        src_addr = utility.array_to_string(device.name)
+        src_port = dst_addr = dst_port = None
+
+        bpfprog = device.xdp_prog
+        if not bpfprog:
+            return
+
+        if not bpfprog.has_member("aux") or not bpfprog.aux:
+            return
+
+        bpfprog_aux = bpfprog.aux
+        if bpfprog_aux.has_member("id"):
+            # `id` member was added to `bpf_prog_aux` in kernels 4.13
+            bpfprog_id = bpfprog_aux.id
+            dst_port = f"ebpf_prog_id:{bpfprog_id}"
+        if bpfprog_aux.has_member("name"):
+            # `name` was added to `bpf_prog_aux` in kernels 4.15
+            bpf_name = utility.array_to_string(bpfprog_aux.name)
+            if bpf_name:
+                dst_addr = f"ebpf_prog_name:{bpf_name}"
+
+        xsk_state = xdp_sock.get_state()
+        state = xsk_state.replace("XSK_", "")
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return xdp_sock, sock_stat
+
+    def _bluetooth_sock(
+        self, sock: objects.StructType
+    ) -> Tuple[objects.StructType, Tuple[str, str, str]]:
+        """Handles the AF_BLUETOOTH socket family
+
+        Args:
+            sock: Kernel generic `sock` object
+
+        Returns:
+            bt_sock: The kernel's `bt_sock` object
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+        """
+        bt_sock = sock.cast("bt_sock")
+
+        def bt_addr(addr):
+            return ":".join(reversed(["%02x" % x for x in addr.b]))
+
+        src_addr = src_port = dst_addr = dst_port = None
+        bt_protocol = bt_sock.get_protocol()
+        if bt_protocol == "HCI":
+            if self._vmlinux.has_type("hci_pinfo"):
+                pinfo = bt_sock.cast("hci_pinfo")
+                if (
+                    pinfo.has_member("hdev")
+                    and self._vmlinux.has_type("hci_dev")
+                    and pinfo.hdev.has_member("dev_name")
+                ):
+                    src_addr = utility.array_to_string(pinfo.hdev.dev_name)
+            else:
+                vollog.log(
+                    constants.LOGLEVEL_V,
+                    "Type definition for 'hci_pinfo' is not available in the symbols",
+                )
+        elif bt_protocol == "L2CAP":
+            if self._vmlinux.has_type("l2cap_pinfo"):
+                pinfo = bt_sock.cast("l2cap_pinfo")
+                src_addr = bt_addr(pinfo.chan.src)
+                dst_addr = bt_addr(pinfo.chan.dst)
+                src_port = pinfo.chan.sport
+                dst_port = pinfo.chan.psm
+            else:
+                vollog.log(
+                    constants.LOGLEVEL_V,
+                    "Type definition for 'l2cap_pinfo' is not available in the symbols",
+                )
+        elif bt_protocol == "RFCOMM":
+            if self._vmlinux.has_type("rfcomm_pinfo"):
+                pinfo = bt_sock.cast("rfcomm_pinfo")
+                src_addr = bt_addr(pinfo.src)
+                dst_addr = bt_addr(pinfo.dst)
+                src_port = pinfo.channel
+            else:
+                vollog.log(
+                    constants.LOGLEVEL_V,
+                    "Type definition for 'rfcomm_pinfo' is not available in the symbols",
+                )
+        elif bt_protocol == "SCO":
+            if self._vmlinux.has_type("sco_pinfo"):
+                pinfo = bt_sock.cast("sco_pinfo")
+                src_addr = bt_addr(pinfo.src)
+                dst_addr = bt_addr(pinfo.dst)
+            else:
+                vollog.log(
+                    constants.LOGLEVEL_V,
+                    "Type definition for 'sco_pinfo' is not available in the symbols",
+                )
+        else:
+            vollog.log(
+                constants.LOGLEVEL_V, "Unsupported bluetooth protocol '%s'", bt_protocol
+            )
+
+        state = bt_sock.get_state()
+
+        sock_stat = src_addr, src_port, dst_addr, dst_port, state
+        return bt_sock, sock_stat
+
+
+class Sockstat(plugins.PluginInterface):
+    """Lists all network connections for all processes."""
+
+    _required_framework_version = (2, 0, 0)
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls):
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.VersionRequirement(
+                name="SockHandlers", component=SockHandlers, version=(1, 0, 0)
+            ),
+            requirements.PluginRequirement(
+                name="lsof", plugin=lsof.Lsof, version=(1, 1, 0)
+            ),
+            requirements.VersionRequirement(
+                name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
+            ),
+            requirements.BooleanRequirement(
+                name="unix",
+                description=("Show UNIX domain Sockets only"),
+                default=False,
+                optional=True,
+            ),
+            requirements.ListRequirement(
+                name="pids",
+                description="Filter results by process IDs. "
+                "It takes the root PID namespace identifiers.",
+                element_type=int,
+                optional=True,
+            ),
+            requirements.IntRequirement(
+                name="netns",
+                description="Filter results by network namespace. "
+                "Otherwise, all of them are shown.",
+                optional=True,
+            ),
+        ]
+
+    @classmethod
+    def list_sockets(
+        cls,
+        context: interfaces.context.ContextInterface,
+        symbol_table: str,
+        filter_func: Callable[[int], bool] = lambda _: False,
+    ):
+        """Returns every single socket descriptor
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            symbol_table: The name of the kernel module on which to operate
+            filter_func: A function which takes a task object and returns True if the task should be ignored/filtered
+
+        Yields:
+            task: Kernel's task object
+            netns_id: Network namespace ID
+            fd_num: File descriptor number
+            family: Socket family string (AF_UNIX, AF_INET, etc)
+            sock_type: Socket type string (STREAM, DGRAM, etc)
+            protocol: Protocol string (UDP, TCP, etc)
+            sock_fields: A tuple with the *_sock object, the sock stats and the
+                         extended info dictionary
+        """
+        vmlinux = context.modules[symbol_table]
+
+        sfop_addr = vmlinux.object_from_symbol("socket_file_ops").vol.offset
+        dfop_addr = vmlinux.object_from_symbol("sockfs_dentry_operations").vol.offset
+
+        fd_generator = lsof.Lsof.list_fds(context, vmlinux.name, filter_func)
+        for _pid, _task_comm, task, fd_fields in fd_generator:
+            fd_num, filp, _full_path = fd_fields
+
+            if filp.f_op not in (sfop_addr, dfop_addr):
+                continue
+
+            dentry = filp.get_dentry()
+            if not dentry:
+                continue
+
+            d_inode = dentry.d_inode
+            if not d_inode:
+                continue
+
+            socket_alloc = linux.LinuxUtilities.container_of(
+                d_inode, "socket_alloc", "vfs_inode", vmlinux
+            )
+            socket = socket_alloc.socket
+
+            if not (socket and socket.sk):
+                continue
+
+            sock = socket.sk.dereference()
+
+            sock_type = sock.get_type()
+            family = sock.get_family()
+
+            sock_handler = SockHandlers(vmlinux, task)
+            sock_fields = sock_handler.process_sock(sock)
+            if not sock_fields:
+                continue
+
+            child_sock = sock_fields[0]
+            protocol = child_sock.get_protocol()
+
+            net = task.nsproxy.net_ns
+            netns_id = net.get_inode()
+            yield task, netns_id, fd_num, family, sock_type, protocol, sock_fields
+
+    def _format_fields(self, sock_stat, protocol):
+        """Prepare the socket fields to be rendered
+
+        Args:
+            sock_stat: A tuple with the source and destination (address and port) along with its state string
+            protocol: Protocol string (UDP, TCP, etc)
+
+        Returns:
+            `sock_stat` and `protocol` formatted.
+        """
+        sock_stat = [
+            NotAvailableValue() if field is None else str(field) for field in sock_stat
+        ]
+        if protocol is None:
+            protocol = NotAvailableValue()
+
+        return tuple(sock_stat), protocol
+
+    def _generator(self, pids: List[int], netns_id_arg: int, symbol_table: str):
+        """Enumerate tasks sockets. Each row represents a kernel socket.
+
+        Args:
+            pids: List of PIDs to filter. If a empty list or
+            netns_id_arg: If a network namespace ID is set, it will only show this namespace.
+            symbol_table: The name of the kernel module on which to operate
+
+        Yields:
+            netns_id: Network namespace ID
+            family: Socket family string (AF_UNIX, AF_INET, etc)
+            sock_type: Socket type string (STREAM, DGRAM, etc)
+            protocol: Protocol string (UDP, TCP, etc)
+            source addr: Source address string
+            source port: Source port string (not all of them are int)
+            destination addr: Destination address string
+            destination port: Destination port (not all of them are int)
+            state: State strings (LISTEN, CONNECTED, etc)
+            tasks: String with a list of tasks and FDs using a socket. It can also have
+                   extended information such as socket filters, bpf info, etc.
+        """
+        filter_func = lsof.pslist.PsList.create_pid_filter(pids)
+        socket_generator = self.list_sockets(
+            self.context, symbol_table, filter_func=filter_func
+        )
+
+        for (
+            task,
+            netns_id,
+            fd_num,
+            family,
+            sock_type,
+            protocol,
+            sock_fields,
+        ) in socket_generator:
+            if netns_id_arg and netns_id_arg != netns_id:
+                continue
+
+            sock, sock_stat, extended = sock_fields
+            sock_stat, protocol = self._format_fields(sock_stat, protocol)
+
+            socket_filter_str = (
+                ",".join(f"{k}={v}" for k, v in extended.items())
+                if extended
+                else NotAvailableValue()
+            )
+
+            fields = (
+                netns_id,
+                task.pid,
+                fd_num,
+                format_hints.Hex(sock.vol.offset),
+                family,
+                sock_type,
+                protocol,
+                *sock_stat,
+                socket_filter_str,
+            )
+
+            yield (0, fields)
+
+    def run(self):
+        pids = self.config.get("pids")
+        netns_id = self.config["netns"]
+        symbol_table = self.config["kernel"]
+
+        tree_grid_args = [
+            ("NetNS", int),
+            ("Pid", int),
+            ("FD", int),
+            ("Sock Offset", format_hints.Hex),
+            ("Family", str),
+            ("Type", str),
+            ("Proto", str),
+            ("Source Addr", str),
+            ("Source Port", str),
+            ("Destination Addr", str),
+            ("Destination Port", str),
+            ("State", str),
+            ("Filter", str),
+        ]
+
+        return TreeGrid(tree_grid_args, self._generator(pids, netns_id, symbol_table))

volatility3/framework/plugins/linux/tty_check.py

@@ -61,7 +61,6 @@ class tty_check(plugins.PluginInterface):
         for tty in tty_drivers.to_list(
             vmlinux.symbol_table_name + constants.BANG + "tty_driver", "tty_drivers"
         ):
-
             try:
                 ttys = utility.array_of_pointers(
                     tty.ttys.dereference(),
@@ -73,7 +72,6 @@ class tty_check(plugins.PluginInterface):
                 continue
 
             for tty_dev in ttys:
-
                 if tty_dev == 0:
                     continue
 

volatility3/framework/plugins/mac/check_syscall.py

@@ -47,7 +47,7 @@ class Check_syscall(plugins.PluginInterface):
 
         table = kernel.object_from_symbol(symbol_name="sysent")
 
-        for (i, ent) in enumerate(table):
+        for i, ent in enumerate(table):
             try:
                 call_addr = ent.sy_call.dereference().vol.offset
             except exceptions.InvalidAddressException:

volatility3/framework/plugins/mac/kauth_listeners.py

@@ -49,7 +49,6 @@ class Kauth_listeners(interfaces.plugins.PluginInterface):
         for scope in kauth_scopes.Kauth_scopes.list_kauth_scopes(
             self.context, self.config["kernel"]
         ):
-
             scope_name = utility.pointer_to_string(scope.ks_identifier, 128)
 
             for listener in scope.get_listeners():

volatility3/framework/plugins/mac/kauth_scopes.py

@@ -65,7 +65,6 @@ class Kauth_scopes(interfaces.plugins.PluginInterface):
         )
 
         for scope in self.list_kauth_scopes(self.context, self.config["kernel"]):
-
             callback = scope.ks_callback
             if callback == 0:
                 continue

volatility3/framework/plugins/mac/kevents.py

@@ -184,7 +184,6 @@ class Kevents(interfaces.plugins.PluginInterface):
         for task_name, pid, kn in self.list_kernel_events(
             self.context, self.config["kernel"], filter_func=filter_func
         ):
-
             filter_index = kn.kn_kevent.filter * -1
             if filter_index in self.event_types:
                 filter_name = self.event_types[filter_index]

volatility3/framework/plugins/mac/list_files.py

@@ -137,7 +137,6 @@ class List_Files(plugins.PluginInterface):
     def _walk_mounts(
         cls, context: interfaces.context.ContextInterface, kernel_module_name: str
     ) -> Iterable[interfaces.objects.ObjectInterface]:
-
         loop_vnodes = {}
 
         # iterate each vnode source from each mount
@@ -186,7 +185,6 @@ class List_Files(plugins.PluginInterface):
     def list_files(
         cls, context: interfaces.context.ContextInterface, kernel_module_name: str
     ) -> Iterable[interfaces.objects.ObjectInterface]:
-
         vnodes = cls._walk_mounts(context, kernel_module_name)
 
         for voff, (vnode_name, parent_offset, vnode) in vnodes.items():
@@ -196,7 +194,6 @@ class List_Files(plugins.PluginInterface):
 
     def _generator(self):
         for vnode, full_path in self.list_files(self.context, self.config["kernel"]):
-
             yield (0, (format_hints.Hex(vnode.vol.offset), full_path))
 
     def run(self):

volatility3/framework/plugins/mac/lsmod.py

@@ -63,7 +63,6 @@ class Lsmod(plugins.PluginInterface):
         seen: Set = set()
 
         while kmod != 0 and kmod not in seen and len(seen) < 1024:
-
             kmod_obj = kmod.dereference()
 
             if not kernel_layer.is_valid(kmod_obj.vol.offset, kmod_obj.vol.size):
@@ -81,7 +80,6 @@ class Lsmod(plugins.PluginInterface):
 
     def _generator(self):
         for module in self.list_modules(self.context, self.config["kernel"]):
-
             mod_name = utility.array_to_string(module.name)
             mod_size = module.size
 

volatility3/framework/plugins/mac/netstat.py

@@ -68,7 +68,6 @@ class Netstat(plugins.PluginInterface):
         # This is hardcoded, since a change in the default method would change the expected results
         list_tasks = pslist.PsList.get_list_tasks(pslist.PsList.pslist_methods[0])
         for task in list_tasks(context, kernel_module_name, filter_func):
-
             task_name = utility.array_to_string(task.p_comm)
             pid = task.p_pid
 
@@ -101,7 +100,6 @@ class Netstat(plugins.PluginInterface):
         for task_name, pid, socket in self.list_sockets(
             self.context, self.config["kernel"], filter_func=filter_func
         ):
-
             family = socket.get_family()
 
             if family == 1:

volatility3/framework/plugins/mac/pslist.py

@@ -83,7 +83,6 @@ class PsList(interfaces.plugins.PluginInterface):
 
     @classmethod
     def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[int], bool]:
-
         filter_func = lambda _: False
         # FIXME: mypy #4973 or #2608
         pid_list = pid_list or []

volatility3/framework/plugins/timeliner.py

@@ -136,7 +136,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
             )
             try:
                 vollog.log(logging.INFO, f"Running {plugin_name}")
-                for (item, timestamp_type, timestamp) in plugin.generate_timeline():
+                for item, timestamp_type, timestamp in plugin.generate_timeline():
                     times = self.timeline.get((plugin_name, item), {})
                     if times.get(timestamp_type, None) is not None:
                         vollog.debug(

volatility3/framework/plugins/windows/bigpools.py

@@ -141,7 +141,6 @@ class BigPools(interfaces.plugins.PluginInterface):
             tags=tags,
             show_free=self.config.get("show-free"),
         ):
-
             num_bytes = big_pool.get_number_of_bytes()
             if not isinstance(num_bytes, interfaces.renderers.BaseAbsentValue):
                 num_bytes = format_hints.Hex(num_bytes)

volatility3/framework/plugins/windows/cachedump.py

@@ -173,7 +173,6 @@ class Cachedump(interfaces.plugins.PluginInterface):
             kernel.symbol_table_name,
             hive_offsets=None if offset is None else [offset],
         ):
-
             if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
                 syshive = hive
             if hive.get_name().split("\\")[-1].upper() == "SECURITY":

volatility3/framework/plugins/windows/callbacks.py

@@ -3,6 +3,7 @@
 #
 
 import logging
+import contextlib
 from typing import List, Iterable, Tuple, Optional, Union
 
 from volatility3.framework import constants, exceptions, renderers, interfaces, symbols
@@ -82,7 +83,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
             context: The context to retrieve required elements (layers, symbol tables) from
             layer_name: The name of the layer on which to operate
             symbol_table: The name of the table containing the kernel symbols
-            callback_table_name: The nae of the table containing the callback symbols
+            callback_table_name: The name of the table containing the callback symbols
 
         Yields:
             A name, location and optional detail string
@@ -103,7 +104,6 @@ class Callbacks(interfaces.plugins.PluginInterface):
         ]
 
         for symbol_name, extended_list in symbol_names:
-
             try:
                 symbol_offset = ntkrnlmp.get_symbol(symbol_name).address
             except exceptions.SymbolError:
@@ -182,7 +182,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
         layer_name: str,
         symbol_table: str,
         callback_table_name: str,
-    ) -> Iterable[Tuple[str, int, None]]:
+    ) -> Iterable[Tuple[str, int, Optional[str]]]:
         """
         Lists all registry callbacks via the CallbackListHead.
         """
@@ -203,7 +203,10 @@ class Callbacks(interfaces.plugins.PluginInterface):
 
         callback_list = ntkrnlmp.object(object_type="_LIST_ENTRY", offset=symbol_offset)
         for callback in callback_list.to_list(full_type_name, "Link"):
-            yield "CmRegisterCallbackEx", callback.Function, f"Altitude: {callback.Altitude.String}"
+            altitude = None
+            with contextlib.suppress(exceptions.InvalidAddressException):
+                altitude = callback.Altitude.String
+            yield "CmRegisterCallbackEx", callback.Function, f"Altitude: {altitude}"
 
     @classmethod
     def list_registry_callbacks(
@@ -212,14 +215,14 @@ class Callbacks(interfaces.plugins.PluginInterface):
         layer_name: str,
         symbol_table: str,
         callback_table_name: str,
-    ) -> Iterable[Tuple[str, int, None]]:
+    ) -> Iterable[Tuple[str, int, Optional[str]]]:
         """Lists all registry callbacks.
 
         Args:
             context: The context to retrieve required elements (layers, symbol tables) from
             layer_name: The name of the layer on which to operate
             symbol_table: The name of the table containing the kernel symbols
-            callback_table_name: The nae of the table containing the callback symbols
+            callback_table_name: The name of the table containing the callback symbols
 
         Yields:
             A name, location and optional detail string
@@ -269,7 +272,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
             context: The context to retrieve required elements (layers, symbol tables) from
             layer_name: The name of the layer on which to operate
             symbol_table: The name of the table containing the kernel symbols
-            callback_table_name: The nae of the table containing the callback symbols
+            callback_table_name: The name of the table containing the callback symbols
 
         Yields:
             A name, location and optional detail string
@@ -327,7 +330,7 @@ class Callbacks(interfaces.plugins.PluginInterface):
             context: The context to retrieve required elements (layers, symbol tables) from
             layer_name: The name of the layer on which to operate
             symbol_table: The name of the table containing the kernel symbols
-            callback_table_name: The nae of the table containing the callback symbols
+            callback_table_name: The name of the table containing the callback symbols
 
         Yields:
             A name, location and optional detail string
@@ -350,7 +353,6 @@ class Callbacks(interfaces.plugins.PluginInterface):
         )
 
         for callback in callback_record.Entry:
-
             if not context.layers[layer_name].is_valid(callback.CallbackRoutine, 64):
                 continue
 
@@ -368,7 +370,6 @@ class Callbacks(interfaces.plugins.PluginInterface):
             yield "KeBugCheckCallbackListHead", callback.CallbackRoutine, component
 
     def _generator(self):
-
         kernel = self.context.modules[self.config["kernel"]]
 
         callback_table_name = self.create_callback_table(
@@ -393,7 +394,6 @@ class Callbacks(interfaces.plugins.PluginInterface):
                 kernel.symbol_table_name,
                 callback_table_name,
             ):
-
                 if callback_detail is None:
                     detail = renderers.NotApplicableValue()
                 else:
@@ -447,7 +447,6 @@ class Callbacks(interfaces.plugins.PluginInterface):
                     )
 
     def run(self):
-
         return renderers.TreeGrid(
             [
                 ("Type", str),

volatility3/framework/plugins/windows/devicetree.py

@@ -180,7 +180,7 @@ class DeviceTree(interfaces.plugins.PluginInterface):
                             ),
                         )
 
-            except (exceptions.InvalidAddressException):
+            except exceptions.InvalidAddressException:
                 vollog.log(
                     constants.LOGLEVEL_VVVV,
                     f"Invalid address identified in drivers and devices: {driver.vol.offset:x}",

volatility3/framework/plugins/windows/dlllist.py

@@ -129,12 +129,10 @@ class DllList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             nt_major_version == 6 and nt_minor_version >= 1
         )
         for proc in procs:
-
             proc_id = proc.UniqueProcessId
             proc_layer_name = proc.add_process_layer()
 
             for entry in proc.load_order_modules():
-
                 BaseDllName = FullDllName = renderers.UnreadableValue()
                 with contextlib.suppress(exceptions.InvalidAddressException):
                     BaseDllName = entry.BaseDllName.get_string()

volatility3/framework/plugins/windows/driverirp.py

@@ -71,7 +71,6 @@ class DriverIrp(interfaces.plugins.PluginInterface):
         for driver in driverscan.DriverScan.scan_drivers(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 driver_name = driver.get_driver_name()
             except (ValueError, exceptions.InvalidAddressException):
@@ -113,7 +112,6 @@ class DriverIrp(interfaces.plugins.PluginInterface):
                         )
 
     def run(self):
-
         return renderers.TreeGrid(
             [
                 ("Offset", format_hints.Hex),

volatility3/framework/plugins/windows/drivermodule.py

@@ -73,7 +73,6 @@ class DriverModule(interfaces.plugins.PluginInterface):
                 )
 
     def run(self) -> renderers.TreeGrid:
-
         return renderers.TreeGrid(
             [
                 ("Offset", format_hints.Hex),

volatility3/framework/plugins/windows/driverscan.py

@@ -54,7 +54,6 @@ class DriverScan(interfaces.plugins.PluginInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 

volatility3/framework/plugins/windows/dumpfiles.py

@@ -229,7 +229,6 @@ class DumpFiles(interfaces.plugins.PluginInterface):
             )
 
             for proc in procs:
-
                 try:
                     object_table = proc.ObjectTable
                 except exceptions.InvalidAddressException:

volatility3/framework/plugins/windows/envars.py

@@ -221,7 +221,6 @@ class Envars(interfaces.plugins.PluginInterface):
                 )
 
     def run(self):
-
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
         kernel = self.context.modules[self.config["kernel"]]
 

volatility3/framework/plugins/windows/filescan.py

@@ -53,7 +53,6 @@ class FileScan(interfaces.plugins.PluginInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 
@@ -63,7 +62,6 @@ class FileScan(interfaces.plugins.PluginInterface):
         for fileobj in self.scan_files(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 file_name = fileobj.FileName.String
             except exceptions.InvalidAddressException:

volatility3/framework/plugins/windows/getservicesids.py

@@ -73,7 +73,6 @@ class GetServiceSIDs(interfaces.plugins.PluginInterface):
         ]
 
     def _generator(self):
-
         kernel = self.context.modules[self.config["kernel"]]
         # Get the system hive
         for hive in hivelist.HiveList.list_hives(

volatility3/framework/plugins/windows/getsids.py

@@ -112,7 +112,6 @@ class GetSIDs(interfaces.plugins.PluginInterface):
             filter_string="config\\software",
             hive_offsets=None,
         ):
-
             try:
                 for subkey in hive.get_key(key).get_subkeys():
                     sid = str(subkey.get_name())
@@ -165,7 +164,6 @@ class GetSIDs(interfaces.plugins.PluginInterface):
         return sids
 
     def _generator(self, procs):
-
         user_sids = self.lookup_user_sids()
 
         # Go all over the process list, get the token
@@ -214,7 +212,6 @@ class GetSIDs(interfaces.plugins.PluginInterface):
                 )
 
     def run(self):
-
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
         kernel = self.context.modules[self.config["kernel"]]
 

volatility3/framework/plugins/windows/handles.py

@@ -136,7 +136,6 @@ class Handles(interfaces.plugins.PluginInterface):
         """
 
         if self._sar_value is None:
-
             if not has_capstone:
                 return None
             kernel = self.context.modules[self.config["kernel"]]
@@ -160,7 +159,7 @@ class Handles(interfaces.plugins.PluginInterface):
 
             md = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64)
 
-            for (address, size, mnemonic, op_str) in md.disasm_lite(
+            for address, size, mnemonic, op_str in md.disasm_lite(
                 data, kvo + func_addr
             ):
                 # print("{} {} {} {}".format(address, size, mnemonic, op_str))
@@ -300,7 +299,6 @@ class Handles(interfaces.plugins.PluginInterface):
         masked_offset = offset & layer_object.maximum_address
 
         for entry in table:
-
             if level > 0:
                 for x in self._make_handle_array(entry, level - 1, depth):
                     yield x
@@ -329,7 +327,6 @@ class Handles(interfaces.plugins.PluginInterface):
                     continue
 
     def handles(self, handle_table):
-
         try:
             TableCode = handle_table.TableCode & ~self._level_mask
             table_levels = handle_table.TableCode & self._level_mask
@@ -395,7 +392,7 @@ class Handles(interfaces.plugins.PluginInterface):
                         except (ValueError, exceptions.InvalidAddressException):
                             obj_name = ""
 
-                except (exceptions.InvalidAddressException):
+                except exceptions.InvalidAddressException:
                     vollog.log(
                         constants.LOGLEVEL_VVV,
                         f"Cannot access _OBJECT_HEADER at {entry.vol.offset:#x}",
@@ -416,7 +413,6 @@ class Handles(interfaces.plugins.PluginInterface):
                 )
 
     def run(self):
-
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
         kernel = self.context.modules[self.config["kernel"]]
 

volatility3/framework/plugins/windows/hashdump.py

@@ -602,7 +602,6 @@ class Hashdump(interfaces.plugins.PluginInterface):
             kernel.symbol_table_name,
             hive_offsets=None if offset is None else [offset],
         ):
-
             if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
                 syshive = hive
             if hive.get_name().split("\\")[-1].upper() == "SAM":

volatility3/framework/plugins/windows/info.py

@@ -187,7 +187,6 @@ class Info(plugins.PluginInterface):
         return nt_header
 
     def _generator(self):
-
         kernel = self.context.modules[self.config["kernel"]]
 
         layer_name = kernel.layer_name
@@ -215,7 +214,6 @@ class Info(plugins.PluginInterface):
             yield (0, (layer.name, f"{i} {layer.__class__.__name__}"))
 
         if kdbg.Header.OwnerTag == 0x4742444B:
-
             yield (0, ("KdDebuggerDataBlock", hex(kdbg.vol.offset)))
             yield (0, ("NTBuildLab", kdbg.get_build_lab()))
             yield (0, ("CSDVersion", str(kdbg.get_csdversion())))
@@ -285,5 +283,4 @@ class Info(plugins.PluginInterface):
         )
 
     def run(self):
-
         return TreeGrid([("Variable", str), ("Value", str)], self._generator())

volatility3/framework/plugins/windows/joblinks.py

@@ -103,7 +103,7 @@ class JobLinks(interfaces.plugins.PluginInterface):
                         ),
                     )
 
-            except (exceptions.InvalidAddressException):
+            except exceptions.InvalidAddressException:
                 continue
 
     def run(self) -> renderers.TreeGrid:

volatility3/framework/plugins/windows/ldrmodules.py

@@ -33,7 +33,6 @@ class LdrModules(interfaces.plugins.PluginInterface):
         ]
 
     def _generator(self, procs):
-
         pe_table_name = intermed.IntermediateSymbolTable.create(
             self.context, self.config_path, "windows", "pe", class_types=pe.class_types
         )

volatility3/framework/plugins/windows/lsadump.py

@@ -118,12 +118,10 @@ class Lsadump(interfaces.plugins.PluginInterface):
         if enc_secret_key:
             enc_secret_value = next(enc_secret_key.get_values())
             if enc_secret_value:
-
                 enc_secret = sechive.read(
                     enc_secret_value.Data + 4, enc_secret_value.DataLength
                 )
                 if enc_secret:
-
                     if not is_vista_or_later:
                         secret = cls.decrypt_secret(enc_secret[0xC:], lsakey)
                     else:
@@ -160,7 +158,6 @@ class Lsadump(interfaces.plugins.PluginInterface):
     def _generator(
         self, syshive: registry.RegistryHive, sechive: registry.RegistryHive
     ):
-
         kernel = self.context.modules[self.config["kernel"]]
 
         vista_or_later = versions.is_vista_or_later(
@@ -183,7 +180,6 @@ class Lsadump(interfaces.plugins.PluginInterface):
             return
 
         for key in secrets_key.get_subkeys():
-
             sec_val_key = hashdump.Hashdump.get_hive_key(
                 sechive,
                 "Policy\\Secrets\\" + key.get_key_path().split("\\")[3] + "\\CurrVal",
@@ -208,7 +204,6 @@ class Lsadump(interfaces.plugins.PluginInterface):
             yield (0, (key.get_name(), secret.decode("latin1"), secret))
 
     def run(self):
-
         offset = self.config.get("offset", None)
         syshive = sechive = None
         kernel = self.context.modules[self.config["kernel"]]
@@ -220,7 +215,6 @@ class Lsadump(interfaces.plugins.PluginInterface):
             kernel.symbol_table_name,
             hive_offsets=None if offset is None else [offset],
         ):
-
             if hive.get_name().split("\\")[-1].upper() == "SYSTEM":
                 syshive = hive
             if hive.get_name().split("\\")[-1].upper() == "SECURITY":

volatility3/framework/plugins/windows/malfind.py

@@ -151,7 +151,6 @@ class Malfind(interfaces.plugins.PluginInterface):
             for vad, data in self.list_injections(
                 self.context, kernel.layer_name, kernel.symbol_table_name, proc
             ):
-
                 # if we're on a 64 bit kernel, we may still need 32 bit disasm due to wow64
                 if is_32bit_arch or proc.get_is_wow64():
                     architecture = "intel"

volatility3/framework/plugins/windows/mbrscan.py

@@ -99,7 +99,6 @@ class MBRScan(interfaces.plugins.PluginInterface):
                     all_zeros = bootcode.count(b"\x00") == len(bootcode)
 
                 if not all_zeros:
-
                     partition_entries = [
                         partition_table.FirstEntry,
                         partition_table.SecondEntry,
@@ -155,7 +154,6 @@ class MBRScan(interfaces.plugins.PluginInterface):
                     for partition_index, partition_entry_object in enumerate(
                         partition_entries, start=1
                     ):
-
                         if not self.config.get("full", True):
                             yield (
                                 1,

volatility3/framework/plugins/windows/modscan.py

@@ -70,7 +70,6 @@ class ModScan(interfaces.plugins.PluginInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 
@@ -175,7 +174,6 @@ class ModScan(interfaces.plugins.PluginInterface):
         for mod in self.scan_modules(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 BaseDllName = mod.BaseDllName.get_string()
             except exceptions.InvalidAddressException:
@@ -188,7 +186,6 @@ class ModScan(interfaces.plugins.PluginInterface):
 
             file_output = "Disabled"
             if self.config["dump"]:
-
                 session_layer_name = self.find_session_layer(
                     self.context, session_layers, mod.DllBase
                 )

volatility3/framework/plugins/windows/modules.py

@@ -53,7 +53,6 @@ class Modules(interfaces.plugins.PluginInterface):
         for mod in self.list_modules(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 BaseDllName = mod.BaseDllName.get_string()
             except exceptions.InvalidAddressException:

volatility3/framework/plugins/windows/mutantscan.py

@@ -53,7 +53,6 @@ class MutantScan(interfaces.plugins.PluginInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 
@@ -63,7 +62,6 @@ class MutantScan(interfaces.plugins.PluginInterface):
         for mutant in self.scan_mutants(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 name = mutant.get_name()
             except (ValueError, exceptions.InvalidAddressException):

volatility3/framework/plugins/windows/netscan.py

@@ -375,7 +375,6 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, nt_symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 
@@ -394,7 +393,6 @@ class NetScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             kernel.symbol_table_name,
             netscan_symbol_table,
         ):
-
             vollog.debug(
                 f"Found netw obj @ 0x{netw_obj.vol.offset:2x} of assumed type {type(netw_obj)}"
             )

volatility3/framework/plugins/windows/netstat.py

@@ -329,7 +329,6 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
                     alignment,
                     net_symbol_table,
                 ):
-
                     endpoint = context.object(
                         obj_name,
                         layer_name=layer_name,
@@ -591,7 +590,6 @@ class NetStat(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             tcpip_module.DllBase,
             tcpip_symbol_table,
         ):
-
             # objects passed pool header constraints. check for additional constraints if strict flag is set.
             if not show_corrupt_results and not netw_obj.is_valid():
                 continue

volatility3/framework/plugins/windows/poolscanner.py

@@ -144,7 +144,6 @@ class PoolScanner(plugins.PluginInterface):
         ]
 
     def _generator(self):
-
         kernel = self.context.modules[self.config["kernel"]]
 
         symbol_table = kernel.symbol_table_name
@@ -367,7 +366,6 @@ class PoolScanner(plugins.PluginInterface):
         for constraint, header in cls.pool_scan(
             context, scan_layer, symbol_table, constraints, alignment=alignment
         ):
-
             mem_objects = header.get_object(
                 constraint=constraint,
                 use_top_down=is_windows_8_or_later,

volatility3/framework/plugins/windows/privileges.py

@@ -66,7 +66,6 @@ class Privs(interfaces.plugins.PluginInterface):
         ]
 
     def _generator(self, procs):
-
         for task in procs:
             try:
                 process_token = task.Token.dereference().cast("_TOKEN")
@@ -107,7 +106,6 @@ class Privs(interfaces.plugins.PluginInterface):
                 )
 
     def run(self):
-
         filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
         kernel = self.context.modules[self.config["kernel"]]
 

volatility3/framework/plugins/windows/pslist.py

@@ -226,7 +226,6 @@ class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             kernel.symbol_table_name,
             filter_func=self.create_pid_filter(self.config.get("pid", None)),
         ):
-
             if not self.config.get("physical", self.PHYSICAL_DEFAULT):
                 offset = proc.vol.offset
             else:

volatility3/framework/plugins/windows/psscan.py

@@ -87,7 +87,6 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             if not filter_func(mem_object):
                 yield mem_object
@@ -192,7 +191,6 @@ class PsScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
             kernel.symbol_table_name,
             filter_func=pslist.PsList.create_pid_filter(self.config.get("pid", None)),
         ):
-
             file_output = "Disabled"
             if self.config["dump"]:
                 # windows 10 objects (maybe others in the future) are already in virtual memory

volatility3/framework/plugins/windows/pstree.py

@@ -3,7 +3,7 @@
 #
 import datetime
 import logging
-from typing import Dict, Set, Tuple
+from typing import Callable, Dict, Set, Tuple
 
 from volatility3.framework import objects, interfaces, renderers
 from volatility3.framework.configuration import requirements
@@ -24,6 +24,7 @@ class PsTree(interfaces.plugins.PluginInterface):
         self._processes: Dict[int, Tuple[interfaces.objects.ObjectInterface, int]] = {}
         self._levels: Dict[int, int] = {}
         self._children: Dict[int, Set[int]] = {}
+        self._ancestors: Set[int] = set([])
 
     @classmethod
     def get_requirements(cls):
@@ -45,18 +46,26 @@ class PsTree(interfaces.plugins.PluginInterface):
             requirements.ListRequirement(
                 name="pid",
                 element_type=int,
-                description="Process ID to include (all other processes are excluded)",
+                description="Process ID to include (with ancestors and descendants, all other processes are excluded)",
                 optional=True,
             ),
         ]
 
-    def find_level(self, pid: objects.Pointer) -> None:
+    def find_level(
+        self,
+        pid: objects.Pointer,
+        filter_func: Callable[
+            [interfaces.objects.ObjectInterface], bool
+        ] = lambda _: False,
+    ) -> None:
         """Finds how deep the pid is in the processes list."""
-        seen = set([])
-        seen.add(pid)
+        seen = {pid}
         level = 0
         proc, _ = self._processes.get(pid, None)
+        filtered = not filter_func(proc)
         while proc is not None and proc.InheritedFromUniqueProcessId not in seen:
+            if filtered:
+                self._ancestors.add(proc.UniqueProcessId)
             child_list = self._children.get(proc.InheritedFromUniqueProcessId, set([]))
             child_list.add(proc.UniqueProcessId)
             self._children[proc.InheritedFromUniqueProcessId] = child_list
@@ -67,7 +76,12 @@ class PsTree(interfaces.plugins.PluginInterface):
             level += 1
         self._levels[pid] = level
 
-    def _generator(self):
+    def _generator(
+        self,
+        filter_func: Callable[
+            [interfaces.objects.ObjectInterface], bool
+        ] = lambda _: False,
+    ):
         """Generates the Tree of processes."""
         kernel = self.context.modules[self.config["kernel"]]
 
@@ -87,15 +101,21 @@ class PsTree(interfaces.plugins.PluginInterface):
 
         # Build the child/level maps
         for pid in self._processes:
-            self.find_level(pid)
+            self.find_level(pid, filter_func)
 
         process_pids = set([])
 
-        def yield_processes(pid):
+        def yield_processes(pid, descendant: bool = False):
             if pid in process_pids:
                 vollog.debug(f"Pid cycle: already processed pid {pid}")
                 return
+
             process_pids.add(pid)
+
+            if pid not in self._ancestors and not descendant:
+                vollog.debug(f"Pid cycle: pid {pid} not in filtered tree")
+                return
+
             proc, offset = self._processes[pid]
             row = (
                 proc.UniqueProcessId,
@@ -114,7 +134,9 @@ class PsTree(interfaces.plugins.PluginInterface):
 
             yield (self._levels[pid] - 1, row)
             for child_pid in self._children.get(pid, []):
-                yield from yield_processes(child_pid)
+                yield from yield_processes(
+                    child_pid, descendant or not filter_func(proc)
+                )
 
         for pid in self._levels:
             if self._levels[pid] == 1:
@@ -140,5 +162,9 @@ class PsTree(interfaces.plugins.PluginInterface):
                 ("CreateTime", datetime.datetime),
                 ("ExitTime", datetime.datetime),
             ],
-            self._generator(),
+            self._generator(
+                filter_func=pslist.PsList.create_pid_filter(
+                    self.config.get("pid", None)
+                ),
+            ),
         )

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

@@ -88,7 +88,6 @@ class HiveList(interfaces.plugins.PluginInterface):
             symbol_table=kernel.symbol_table_name,
             filter_string=self.config.get("filter", None),
         ):
-
             file_output = "Disabled"
             if self.config["dump"]:
                 # Construct the hive

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

@@ -86,7 +86,6 @@ class HiveScan(interfaces.plugins.PluginInterface):
         for hive in self.scan_hives(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             yield (0, (format_hints.Hex(hive.vol.offset),))
 
     def run(self):

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

@@ -241,7 +241,6 @@ class PrintKey(interfaces.plugins.PluginInterface):
         key: str = None,
         recurse: bool = False,
     ):
-
         for hive in hivelist.HiveList.list_hives(
             self.context,
             self.config_path,
@@ -249,14 +248,13 @@ class PrintKey(interfaces.plugins.PluginInterface):
             symbol_table=symbol_table,
             hive_offsets=hive_offsets,
         ):
-
             try:
                 # Walk it
                 if key is not None:
                     node_path = hive.get_key(key, return_list=True)
                 else:
                     node_path = [hive.get_node(hive.root_cell_offset)]
-                for (x, y) in self._printkey_iterator(hive, node_path, recurse=recurse):
+                for x, y in self._printkey_iterator(hive, node_path, recurse=recurse):
                     yield (x - len(node_path), y)
             except (
                 exceptions.InvalidAddressException,

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

@@ -248,7 +248,6 @@ class UserAssist(interfaces.plugins.PluginInterface):
 
                 # output any values under Count
                 for value in countkey.get_values():
-
                     value_name = value.get_name()
                     with contextlib.suppress(UnicodeDecodeError):
                         value_name = codecs.encode(value_name, "rot_13")
@@ -281,7 +280,6 @@ class UserAssist(interfaces.plugins.PluginInterface):
                     yield result
 
     def _generator(self):
-
         hive_offsets = None
         if self.config.get("offset", None) is not None:
             hive_offsets = [self.config.get("offset", None)]

volatility3/framework/plugins/windows/sessions.py

@@ -51,7 +51,6 @@ class Sessions(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
             kernel.symbol_table_name,
             filter_func=filter_func,
         ):
-
             session_id = proc.get_session_id()
 
             # Detect RDP, Console or set default value
@@ -112,7 +111,6 @@ class Sessions(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface)
                 yield (description, timeliner.TimeLinerType.CREATED, row_data[5])
 
     def run(self):
-
         return renderers.TreeGrid(
             [
                 ("Session ID", int),

volatility3/framework/plugins/windows/skeleton_key_check.py

@@ -187,7 +187,6 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
         proc_layer_name: str,
         cryptdll_base: int,
     ) -> Tuple[interfaces.objects.ObjectInterface, int, int, int]:
-
         """
         Finds the CSystems array through use of PDB symbols
 
@@ -574,7 +573,6 @@ class Skeleton_Key_Check(interfaces.plugins.PluginInterface):
             scanners.BytesScanner(b"\x17\x00\x00\x00\x01\x00\x00\x00"),
             sections=[(cryptdll_base, cryptdll_size)],
         ):
-
             # this occurs across page boundaries
             if not proc_layer.is_valid(address, ecrypt_size):
                 continue

volatility3/framework/plugins/windows/ssdt.py

@@ -56,7 +56,6 @@ class SSDT(plugins.PluginInterface):
         context_modules = []
 
         for mod in mods:
-
             try:
                 module_name_with_ext = mod.BaseDllName.get_string()
             except exceptions.InvalidAddressException:
@@ -83,7 +82,6 @@ class SSDT(plugins.PluginInterface):
         return contexts.ModuleCollection(context_modules)
 
     def _generator(self) -> Iterator[Tuple[int, Tuple[int, int, Any, Any]]]:
-
         kernel = self.context.modules[self.config["kernel"]]
 
         layer_name = kernel.layer_name
@@ -132,7 +130,6 @@ class SSDT(plugins.PluginInterface):
         )
 
         for idx, function_obj in enumerate(functions):
-
             function = find_address(function_obj)
             module_symbols = collection.get_module_symbols_by_absolute_location(
                 function

volatility3/framework/plugins/windows/svcscan.py

@@ -180,7 +180,6 @@ class SvcScan(interfaces.plugins.PluginInterface):
             symbol_table=kernel.symbol_table_name,
             filter_func=filter_func,
         ):
-
             proc_id = "Unknown"
             try:
                 proc_id = task.UniqueProcessId
@@ -200,7 +199,6 @@ class SvcScan(interfaces.plugins.PluginInterface):
                 scanner=scanners.BytesScanner(needle=service_tag),
                 sections=vadyarascan.VadYaraScan.get_vad_maps(task),
             ):
-
                 if not is_vista_or_later:
                     service_record = self.context.object(
                         service_table_name + constants.BANG + "_SERVICE_RECORD",

volatility3/framework/plugins/windows/symlinkscan.py

@@ -52,7 +52,6 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
         for result in poolscanner.PoolScanner.generate_pool_scan(
             context, layer_name, symbol_table, constraints
         ):
-
             _constraint, mem_object, _header = result
             yield mem_object
 
@@ -62,7 +61,6 @@ class SymlinkScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterfa
         for link in self.scan_symlinks(
             self.context, kernel.layer_name, kernel.symbol_table_name
         ):
-
             try:
                 from_name = link.get_link_name()
             except (ValueError, exceptions.InvalidAddressException):

volatility3/framework/plugins/windows/vadinfo.py

@@ -54,8 +54,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
             requirements.IntRequirement(
                 name="address",
                 description="Process virtual memory address to include "
-                "(all other address ranges are excluded). This must be "
-                "a base address, not an address within the desired range.",
+                "(all other address ranges are excluded).",
                 optional=True,
             ),
             requirements.ListRequirement(
@@ -207,7 +206,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
         if self.config.get("address", None) is not None:
 
             def filter_function(x: interfaces.objects.ObjectInterface) -> bool:
-                return x.get_start() not in [self.config["address"]]
+                return not (x.get_start() <= self.config["address"] <= x.get_end())
 
             filter_func = filter_function
 
@@ -215,7 +214,6 @@ class VadInfo(interfaces.plugins.PluginInterface):
             process_name = utility.array_to_string(proc.ImageFileName)
 
             for vad in self.list_vads(proc, filter_func=filter_func):
-
                 file_output = "Disabled"
                 if self.config["dump"]:
                     file_handle = self.vad_dump(

volatility3/framework/plugins/windows/verinfo.py

@@ -222,7 +222,6 @@ class VerInfo(interfaces.plugins.PluginInterface):
                 continue
 
             for entry in proc.load_order_modules():
-
                 try:
                     BaseDllName = entry.BaseDllName.get_string()
                 except exceptions.InvalidAddressException:

volatility3/framework/plugins/windows/virtmap.py

@@ -31,7 +31,7 @@ class VirtMap(interfaces.plugins.PluginInterface):
 
     def _generator(self, map):
         for entry in sorted(map):
-            for (start, end) in map[entry]:
+            for start, end in map[entry]:
                 yield (0, (entry, format_hints.Hex(start), format_hints.Hex(end)))
 
     @classmethod

volatility3/framework/renderers/__init__.py

@@ -181,7 +181,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
         converted_columns: List[interfaces.renderers.Column] = []
         if len(columns) < 1:
             raise ValueError("Columns must be a list containing at least one column")
-        for (name, column_type) in columns:
+        for name, column_type in columns:
             is_simple_type = issubclass(column_type, self.base_types)
             if not is_simple_type:
                 raise TypeError(
@@ -238,7 +238,7 @@ class TreeGrid(interfaces.renderers.TreeGrid):
         if not self.populated:
             try:
                 prev_nodes: List[interfaces.renderers.TreeNode] = []
-                for (level, item) in self._generator:
+                for level, item in self._generator:
                     parent_index = min(len(prev_nodes), level)
                     parent = prev_nodes[parent_index - 1] if parent_index > 0 else None
                     treenode = self._append(parent, item)

volatility3/framework/renderers/format_hints.py

@@ -36,7 +36,6 @@ class MultiTypeData(bytes):
         split_nulls: bool = False,
         show_hex: bool = False,
     ) -> "MultiTypeData":
-
         if isinstance(original, int):
             data = str(original).encode(encoding)
         else:

volatility3/framework/symbols/__init__.py

@@ -192,7 +192,7 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
                         replacements.add((traverser, child))
                     elif child.children:
                         template_traverse_list.append(child)
-        for (parent, child) in replacements:
+        for parent, child in replacements:
             parent.replace_child(child, self._resolved[child.vol.type_name])
 
     def get_type(self, type_name: str) -> interfaces.objects.Template:

volatility3/framework/symbols/linux/__init__.py

@@ -1,7 +1,7 @@
 # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
-from typing import Iterator, List, Tuple
+from typing import Iterator, List, Tuple, Optional
 
 from volatility3 import framework
 from volatility3.framework import constants, exceptions, interfaces, objects
@@ -27,17 +27,28 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
         self.set_type_class("dentry", extensions.dentry)
         self.set_type_class("fs_struct", extensions.fs_struct)
         self.set_type_class("files_struct", extensions.files_struct)
-        self.set_type_class("vfsmount", extensions.vfsmount)
         self.set_type_class("kobject", extensions.kobject)
+        # Might not exist in the current symbols
+        self.optional_set_type_class("module", extensions.module)
 
-        if "mnt_namespace" in self.types:
-            self.set_type_class("mnt_namespace", extensions.mnt_namespace)
+        # Mount
+        self.set_type_class("vfsmount", extensions.vfsmount)
+        # Might not exist in older kernels or the current symbols
+        self.optional_set_type_class("mount", extensions.mount)
+        self.optional_set_type_class("mnt_namespace", extensions.mnt_namespace)
 
-        if "module" in self.types:
-            self.set_type_class("module", extensions.module)
-
-        if "mount" in self.types:
-            self.set_type_class("mount", extensions.mount)
+        # Network
+        self.set_type_class("net", extensions.net)
+        self.set_type_class("socket", extensions.socket)
+        self.set_type_class("sock", extensions.sock)
+        self.set_type_class("inet_sock", extensions.inet_sock)
+        self.set_type_class("unix_sock", extensions.unix_sock)
+        # Might not exist in older kernels or the current symbols
+        self.optional_set_type_class("netlink_sock", extensions.netlink_sock)
+        self.optional_set_type_class("vsock_sock", extensions.vsock_sock)
+        self.optional_set_type_class("packet_sock", extensions.packet_sock)
+        self.optional_set_type_class("bt_sock", extensions.bt_sock)
+        self.optional_set_type_class("xdp_sock", extensions.xdp_sock)
 
 
 class LinuxUtilities(interfaces.configuration.VersionableInterface):
@@ -51,7 +62,6 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
     # based on __d_path from the Linux kernel
     @classmethod
     def _do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> str:
-
         ret_path: List[str] = []
 
         while dentry != rdentry or vfsmnt != rmnt:
@@ -193,6 +203,9 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
         symbol_table: str,
         task: interfaces.objects.ObjectInterface,
     ):
+        # task.files can be null
+        if not task.files:
+            return
 
         fd_table = task.files.get_fds()
         if fd_table == 0:
@@ -210,7 +223,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
             fd_table, count=max_fds, subtype=file_type, context=context
         )
 
-        for (fd_num, filp) in enumerate(fds):
+        for fd_num, filp in enumerate(fds):
             if filp != 0:
                 full_path = LinuxUtilities.path_for_file(context, task, filp)
 
@@ -304,3 +317,34 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
             )
             yield list_struct
             list_start = getattr(list_struct, list_member)
+
+    @classmethod
+    def container_of(
+        cls,
+        addr: int,
+        type_name: str,
+        member_name: str,
+        vmlinux: interfaces.context.ModuleInterface,
+    ) -> Optional[interfaces.objects.ObjectInterface]:
+        """Cast a member of a structure out to the containing structure.
+        It mimicks the Linux kernel macro container_of() see include/linux.kernel.h
+
+        Args:
+            addr: The pointer to the member.
+            type_name: The type of the container struct this is embedded in.
+            member_name: The name of the member within the struct.
+            vmlinux: The kernel symbols object
+
+        Returns:
+            The constructed object or None
+        """
+
+        if not addr:
+            return
+
+        type_dec = vmlinux.get_type(type_name)
+        member_offset = type_dec.relative_child_offset(member_name)
+        container_addr = addr - member_offset
+        return vmlinux.object(
+            object_type=type_name, offset=container_addr, absolute=True
+        )

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

@@ -4,14 +4,19 @@
 
 import collections.abc
 import logging
+import socket as socket_module
 from typing import Generator, Iterable, Iterator, Optional, Tuple
 
 from volatility3.framework import constants
+from volatility3.framework.constants.linux import SOCK_TYPES, SOCK_FAMILY
+from volatility3.framework.constants.linux import IP_PROTOCOLS, IPV6_PROTOCOLS
+from volatility3.framework.constants.linux import TCP_STATES, NETLINK_PROTOCOLS
+from volatility3.framework.constants.linux import ETH_PROTOCOLS, BLUETOOTH_STATES
+from volatility3.framework.constants.linux import BLUETOOTH_PROTOCOLS, SOCKET_STATES
 from volatility3.framework import exceptions, objects, interfaces, symbols
 from volatility3.framework.layers import linear
 from volatility3.framework.objects import utility
-from volatility3.framework.symbols import generic, linux
-from volatility3.framework.symbols import intermed
+from volatility3.framework.symbols import generic, linux, intermed
 from volatility3.framework.symbols.linux.extensions import elf
 
 vollog = logging.getLogger(__name__)
@@ -590,7 +595,6 @@ class list_head(objects.StructType, collections.abc.Iterable):
 
         seen = {self.vol.offset}
         while link.vol.offset not in seen:
-
             obj = self._context.object(
                 symbol_type, layer, offset=link.vol.offset - relative_offset
             )
@@ -625,7 +629,6 @@ class files_struct(objects.StructType):
 
 
 class mount(objects.StructType):
-
     MNT_NOSUID = 0x01
     MNT_NODEV = 0x02
     MNT_NOEXEC = 0x04
@@ -750,7 +753,6 @@ class mount(objects.StructType):
             and current_mnt.has_parent()
             and current_mnt.vol.offset not in mnt_seen
         ):
-
             current_dentry = current_mnt.mnt_mountpoint
             mnt_seen.add(current_mnt.vol.offset)
             current_mnt = current_mnt.mnt_parent
@@ -837,3 +839,265 @@ class mnt_namespace(objects.StructType):
 
         for mount in self.list.to_list(mnt_type, "mnt_list"):
             yield mount
+
+
+class net(objects.StructType):
+    def get_inode(self):
+        if self.has_member("proc_inum"):
+            return self.proc_inum
+        elif self.ns.has_member("inum"):
+            return self.ns.inum
+        else:
+            raise AttributeError("Unable to find net_namespace inode")
+
+
+class socket(objects.StructType):
+    def _get_vol_kernel(self):
+        symbol_table_arr = self.vol.type_name.split("!", 1)
+        symbol_table = symbol_table_arr[0] if len(symbol_table_arr) == 2 else None
+
+        module_names = list(
+            self._context.modules.get_modules_by_symbol_tables(symbol_table)
+        )
+        if not module_names:
+            raise ValueError(f"No module using the symbol table {symbol_table}")
+
+        kernel_module_name = module_names[0]
+        kernel = self._context.modules[kernel_module_name]
+        return kernel
+
+    def get_inode(self):
+        try:
+            kernel = self._get_vol_kernel()
+        except ValueError:
+            return 0
+
+        socket_alloc = linux.LinuxUtilities.container_of(
+            self.vol.offset, "socket_alloc", "socket", kernel
+        )
+        vfs_inode = socket_alloc.vfs_inode
+
+        return vfs_inode.i_ino
+
+    def get_state(self):
+        socket_state_idx = self.state
+        if 0 <= socket_state_idx < len(SOCKET_STATES):
+            return SOCKET_STATES[socket_state_idx]
+
+
+class sock(objects.StructType):
+    def get_family(self):
+        family_idx = self.__sk_common.skc_family
+        if 0 <= family_idx < len(SOCK_FAMILY):
+            return SOCK_FAMILY[family_idx]
+
+    def get_type(self):
+        return SOCK_TYPES.get(self.sk_type, "")
+
+    def get_inode(self):
+        if not self.sk_socket:
+            return 0
+
+        return self.sk_socket.get_inode()
+
+    def get_protocol(self):
+        return
+
+    def get_state(self):
+        # Return the generic socket state
+        if self.has_member("sk"):
+            return self.sk.sk_socket.get_state()
+
+        return self.sk_socket.get_state()
+
+
+class unix_sock(objects.StructType):
+    def get_name(self):
+        if not self.addr:
+            return
+
+        sockaddr_un = self.addr.name.cast("sockaddr_un")
+        saddr = str(utility.array_to_string(sockaddr_un.sun_path))
+        return saddr
+
+    def get_protocol(self):
+        return
+
+    def get_state(self):
+        """Return a string representing the sock state."""
+
+        # Unix socket states reuse (a subset) of the inet_sock states contants
+        if self.sk.get_type() == "STREAM":
+            state_idx = self.sk.__sk_common.skc_state
+            if 0 <= state_idx < len(TCP_STATES):
+                return TCP_STATES[state_idx]
+        else:
+            # Return the generic socket state
+            return self.sk.sk_socket.get_state()
+
+    def get_inode(self):
+        return self.sk.get_inode()
+
+
+class inet_sock(objects.StructType):
+    def get_family(self):
+        family_idx = self.sk.__sk_common.skc_family
+        if 0 <= family_idx < len(SOCK_FAMILY):
+            return SOCK_FAMILY[family_idx]
+
+    def get_protocol(self):
+        # If INET6 family and a proto is defined, we use that specific IPv6 protocol.
+        # Otherwise, we use the standard IP protocol.
+        protocol = IP_PROTOCOLS.get(self.sk.sk_protocol)
+        if self.get_family() == "AF_INET6":
+            protocol = IPV6_PROTOCOLS.get(self.sk.sk_protocol, protocol)
+
+        return protocol
+
+    def get_state(self):
+        """Return a string representing the sock state."""
+
+        if self.sk.get_type() == "STREAM":
+            state_idx = self.sk.__sk_common.skc_state
+            if 0 <= state_idx < len(TCP_STATES):
+                return TCP_STATES[state_idx]
+        else:
+            # Return the generic socket state
+            return self.sk.sk_socket.get_state()
+
+    def get_src_port(self):
+        sport_le = getattr(self, "sport", getattr(self, "inet_sport", None))
+        if sport_le is not None:
+            return socket_module.htons(sport_le)
+
+    def get_dst_port(self):
+        sk_common = self.sk.__sk_common
+        if hasattr(sk_common, "skc_portpair"):
+            dport_le = sk_common.skc_portpair & 0xFFFF
+        elif hasattr(self, "dport"):
+            dport_le = self.dport
+        elif hasattr(self, "inet_dport"):
+            dport_le = self.inet_dport
+        elif hasattr(sk_common, "skc_dport"):
+            dport_le = sk_common.skc_dport
+        else:
+            return
+
+        return socket_module.htons(dport_le)
+
+    def get_src_addr(self):
+        sk_common = self.sk.__sk_common
+        family = sk_common.skc_family
+        if family == socket_module.AF_INET:
+            addr_size = 4
+            if hasattr(self, "rcv_saddr"):
+                saddr = self.rcv_saddr
+            elif hasattr(self, "inet_rcv_saddr"):
+                saddr = self.inet_rcv_saddr
+            else:
+                saddr = sk_common.skc_rcv_saddr
+        elif family == socket_module.AF_INET6:
+            addr_size = 16
+            saddr = self.pinet6.saddr
+        else:
+            return
+
+        parent_layer = self._context.layers[self.vol.layer_name]
+        try:
+            addr_bytes = parent_layer.read(saddr.vol.offset, addr_size)
+        except exceptions.InvalidAddressException:
+            vollog.debug(
+                f"Unable to read socket src address from {saddr.vol.offset:#x}"
+            )
+            return
+
+        return socket_module.inet_ntop(family, addr_bytes)
+
+    def get_dst_addr(self):
+        sk_common = self.sk.__sk_common
+        family = sk_common.skc_family
+        if family == socket_module.AF_INET:
+            if hasattr(self, "daddr") and self.daddr:
+                daddr = self.daddr
+            elif hasattr(self, "inet_daddr") and self.inet_daddr:
+                daddr = self.inet_daddr
+            else:
+                daddr = sk_common.skc_daddr
+            addr_size = 4
+        elif family == socket_module.AF_INET6:
+            if hasattr(self.pinet6, "daddr"):
+                daddr = self.pinet6.daddr
+            else:
+                daddr = sk_common.skc_v6_daddr
+            addr_size = 16
+        else:
+            return
+
+        parent_layer = self._context.layers[self.vol.layer_name]
+        try:
+            addr_bytes = parent_layer.read(daddr.vol.offset, addr_size)
+        except exceptions.InvalidAddressException:
+            vollog.debug(
+                f"Unable to read socket dst address from {daddr.vol.offset:#x}"
+            )
+            return
+
+        return socket_module.inet_ntop(family, addr_bytes)
+
+
+class netlink_sock(objects.StructType):
+    def get_protocol(self):
+        protocol_idx = self.sk.sk_protocol
+        if 0 <= protocol_idx < len(NETLINK_PROTOCOLS):
+            return NETLINK_PROTOCOLS[protocol_idx]
+
+    def get_state(self):
+        # Return the generic socket state
+        return self.sk.sk_socket.get_state()
+
+
+class vsock_sock(objects.StructType):
+    def get_protocol(self):
+        # The protocol should always be 0 for vsocks
+        return
+
+    def get_state(self):
+        # Return the generic socket state
+        return self.sk.sk_socket.get_state()
+
+
+class packet_sock(objects.StructType):
+    def get_protocol(self):
+        eth_proto = socket_module.htons(self.num)
+        if eth_proto == 0:
+            return
+        elif eth_proto in ETH_PROTOCOLS:
+            return ETH_PROTOCOLS[eth_proto]
+        else:
+            return f"0x{eth_proto:x}"
+
+    def get_state(self):
+        # Return the generic socket state
+        return self.sk.sk_socket.get_state()
+
+
+class bt_sock(objects.StructType):
+    def get_protocol(self):
+        type_idx = self.sk.sk_protocol
+        if 0 <= type_idx < len(BLUETOOTH_PROTOCOLS):
+            return BLUETOOTH_PROTOCOLS[type_idx]
+
+    def get_state(self):
+        state_idx = self.sk.__sk_common.skc_state
+        if 0 <= state_idx < len(BLUETOOTH_STATES):
+            return BLUETOOTH_STATES[state_idx]
+
+
+class xdp_sock(objects.StructType):
+    def get_protocol(self):
+        # The protocol should always be 0 for xdp_sock
+        return
+
+    def get_state(self):
+        # xdp_sock.state is an enum
+        return self.state.lookup()

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

@@ -22,7 +22,6 @@ class elf(objects.StructType):
         size: int,
         members: Dict[str, Tuple[int, interfaces.objects.Template]],
     ) -> None:
-
         super().__init__(
             context=context,
             type_name=type_name,

volatility3/framework/symbols/mac/__init__.py

@@ -70,7 +70,6 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
         kernel,  # ikelos - how to type this??
         mods_list: Iterator[Any],
     ):
-
         try:
             start_addr = kernel.object_from_symbol("vm_kernel_stext")
         except exceptions.SymbolError:
@@ -231,7 +230,6 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
         next_member: str,
         max_elements: int = 4096,
     ) -> Iterable[interfaces.objects.ObjectInterface]:
-
         for element in cls._walk_iterable(
             queue, "tqh_first", "tqe_next", next_member, max_elements
         ):
@@ -244,7 +242,6 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
         next_member: str,
         max_elements: int = 4096,
     ) -> Iterable[interfaces.objects.ObjectInterface]:
-
         for element in cls._walk_iterable(
             queue, "lh_first", "le_next", next_member, max_elements
         ):
@@ -257,7 +254,6 @@ class MacUtilities(interfaces.configuration.VersionableInterface):
         next_member: str,
         max_elements: int = 4096,
     ) -> Iterable[interfaces.objects.ObjectInterface]:
-
         for element in cls._walk_iterable(
             queue, "slh_first", "sle_next", next_member, max_elements
         ):

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

@@ -206,7 +206,7 @@ class vm_map_entry(objects.StructType):
         permask = "rwx"
         perms = ""
 
-        for (ctr, i) in enumerate([1, 3, 5]):
+        for ctr, i in enumerate([1, 3, 5]):
             if (self.protection & i) == i:
                 perms = perms + permask[ctr]
             else:
@@ -593,7 +593,7 @@ class sysctl_oid(objects.StructType):
         checks = [0x80000000, 0x40000000, 0x00800000]
         perms = ["R", "W", "L"]
 
-        for (i, c) in enumerate(checks):
+        for i, c in enumerate(checks):
             if c & self.oid_kind:
                 ret = ret + perms[i]
             else:

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

@@ -202,7 +202,6 @@ class MMVAD_SHORT(objects.StructType):
 
         # this is for windows 8 and 10
         elif self.has_member("VadNode"):
-
             if self.VadNode.has_member("u1"):
                 return self.VadNode.u1.Parent & ~0x3
 
@@ -211,7 +210,6 @@ class MMVAD_SHORT(objects.StructType):
 
         # also for windows 8 and 10
         elif self.has_member("Core"):
-
             if self.Core.VadNode.has_member("u1"):
                 return self.Core.VadNode.u1.Parent & ~0x3
 
@@ -224,14 +222,12 @@ class MMVAD_SHORT(objects.StructType):
         """Get the VAD's starting virtual address. This is the first accessible byte in the range."""
 
         if self.has_member("StartingVpn"):
-
             if self.has_member("StartingVpnHigh"):
                 return (self.StartingVpn << 12) | (self.StartingVpnHigh << 44)
             else:
                 return self.StartingVpn << 12
 
         elif self.has_member("Core"):
-
             if self.Core.has_member("StartingVpnHigh"):
                 return (self.Core.StartingVpn << 12) | (self.Core.StartingVpnHigh << 44)
             else:
@@ -243,7 +239,6 @@ class MMVAD_SHORT(objects.StructType):
         """Get the VAD's ending virtual address. This is the last accessible byte in the range."""
 
         if self.has_member("EndingVpn"):
-
             if self.has_member("EndingVpnHigh"):
                 return (((self.EndingVpn + 1) << 12) | (self.EndingVpnHigh << 44)) - 1
             else:
@@ -376,7 +371,6 @@ class EX_FAST_REF(objects.StructType):
     """
 
     def dereference(self) -> interfaces.objects.ObjectInterface:
-
         if constants.BANG not in self.vol.type_name:
             raise ValueError(
                 f"Invalid symbol table name syntax (no {constants.BANG} found)"
@@ -771,7 +765,6 @@ class EPROCESS(generic.GenericIntelProcess, pool.ExecutiveObject):
         return False
 
     def get_vad_root(self):
-
         # windows 8 and 2012 (_MM_AVL_TABLE)
         if self.VadRoot.has_member("BalancedRoot"):
             return self.VadRoot.BalancedRoot
@@ -1346,7 +1339,6 @@ class SHARED_CACHE_MAP(objects.StructType):
         limit_depth = level_depth
 
         if section_size > self.VACB_SIZE_OF_FIRST_LEVEL:
-
             # Create an array of 128 entries for the VACB index array.
             vacb_array = self._context.object(
                 object_type=symbol_table_name + constants.BANG + "array",

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

@@ -64,7 +64,6 @@ class _TCP_LISTENER(objects.StructType):
         size: int,
         members: Dict[str, Tuple[int, interfaces.objects.Template]],
     ) -> None:
-
         super().__init__(
             context=context,
             type_name=type_name,
@@ -167,7 +166,6 @@ class _TCP_LISTENER(objects.StructType):
                 yield "v6", inaddr6_any, inaddr6_any
 
     def is_valid(self):
-
         try:
             if not self.get_address_family() in (AF_INET, AF_INET6):
                 vollog.debug(
@@ -189,7 +187,6 @@ class _TCP_ENDPOINT(_TCP_LISTENER):
     """Class for objects found in TcpE pools"""
 
     def _ipv4_or_ipv6(self, inaddr):
-
         if self.get_address_family() == AF_INET:
             return inet_ntop(socket.AF_INET, inaddr.addr4)
         else:
@@ -214,7 +211,6 @@ class _TCP_ENDPOINT(_TCP_LISTENER):
             return None
 
     def is_valid(self):
-
         if self.State not in self.State.choices.values():
             vollog.debug(
                 f"{type(self)} 0x{self.vol.offset:x} invalid due to invalid tcp state {self.State}"

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

@@ -151,7 +151,6 @@ class IMAGE_DOS_HEADER(objects.StructType):
 
         counter = 0
         for sect in nt_header.get_sections():
-
             if sect.VirtualAddress > size_of_image:
                 raise ValueError(
                     f"Section VirtualAddress is too large: {sect.VirtualAddress}"

volatility3/framework/symbols/windows/pdbutil.py

@@ -54,6 +54,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
         """
         result = cls.get_guid_from_mz(context, layer_name, offset)
         if result is None:
+            vollog.debug(f"Could not get GUID for {hex(offset)}")
             return None
         guid, age, pdb_name = result
         if config_path is None:
@@ -248,7 +249,6 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
         # Check for writability
         filter_string = os.path.join(pdb_name, guid + "-" + str(age))
         for path in symbols.__path__:
-
             # Store any temporary files created by downloading PDB files
             tmp_files = []
             potential_output_filename = os.path.join(
@@ -352,7 +352,7 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
         if end is None:
             end = ctx.layers[layer_name].maximum_address
 
-        for (GUID, age, pdb_name, signature_offset) in ctx.layers[layer_name].scan(
+        for GUID, age, pdb_name, signature_offset in ctx.layers[layer_name].scan(
             ctx,
             PdbSignatureScanner(pdb_names),
             progress_callback=progress_callback,
@@ -425,7 +425,6 @@ class PDBUtility(interfaces.configuration.VersionableInterface):
         module_size: int = None,
         create_module: bool = False,
     ) -> Tuple[Optional[str], Optional[str]]:
-
         if module_offset is None:
             module_offset = context.layers[layer_name].minimum_address
         if module_size is None: