Linux: Extensions: Clean up the Linux constants imports in the object extension file

Author: gcmoreira

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

@@ -13,24 +13,7 @@
 
 from volatility3.framework import constants, exceptions, objects, interfaces, symbols
 from volatility3.framework.renderers import conversion
-from volatility3.framework.constants.linux import (
-    SOCK_TYPES,
-    SOCK_FAMILY,
-    IP_PROTOCOLS,
-    IPV6_PROTOCOLS,
-    TCP_STATES,
-    NETLINK_PROTOCOLS,
-    ETH_PROTOCOLS,
-    BLUETOOTH_STATES,
-    BLUETOOTH_PROTOCOLS,
-    SOCKET_STATES,
-    CAPABILITIES,
-    PT_FLAGS,
-    MODULE_MAXIMUM_CORE_SIZE,
-    MODULE_MAXIMUM_CORE_TEXT_SIZE,
-    MODULE_MINIMUM_SIZE,
-)
-
+from volatility3.framework.constants import linux as linux_constants
 from volatility3.framework.layers import linear
 from volatility3.framework.objects import utility
 from volatility3.framework.symbols import generic, linux, intermed
@@ -62,9 +45,9 @@ def is_valid(self):
         core_text_size = self.get_core_text_size()
         init_size = self.get_init_size()
         if not (
-            0 < core_text_size <= MODULE_MAXIMUM_CORE_TEXT_SIZE
-            and 0 < core_size <= MODULE_MAXIMUM_CORE_SIZE
-            and core_size + init_size >= MODULE_MINIMUM_SIZE
+            0 < core_text_size <= linux_constants.MODULE_MAXIMUM_CORE_TEXT_SIZE
+            and 0 < core_size <= linux_constants.MODULE_MAXIMUM_CORE_SIZE
+            and core_size + init_size >= linux_constants.MODULE_MINIMUM_SIZE
         ):
             return False
 
@@ -383,7 +366,7 @@ def is_kernel_thread(self) -> bool:
         Returns:
             bool: True, if this task is a kernel thread. Otherwise, False.
         """
-        return (self.flags & constants.linux.PF_KTHREAD) != 0
+        return (self.flags & linux_constants.PF_KTHREAD) != 0
 
     @property
     def is_thread_group_leader(self) -> bool:
@@ -460,7 +443,11 @@ def get_ptrace_tracee_tids(self) -> List[int]:
 
     def get_ptrace_tracee_flags(self) -> Optional[str]:
         """Returns a string with the ptrace flags"""
-        return PT_FLAGS(self.ptrace).flags if self.is_being_ptraced else None
+        return (
+            linux_constants.PT_FLAGS(self.ptrace).flags
+            if self.is_being_ptraced
+            else None
+        )
 
 
 class fs_struct(objects.StructType):
@@ -1567,18 +1554,18 @@ def get_inode(self):
 
     def get_state(self):
         socket_state_idx = self.state
-        if 0 <= socket_state_idx < len(SOCKET_STATES):
-            return SOCKET_STATES[socket_state_idx]
+        if 0 <= socket_state_idx < len(linux_constants.SOCKET_STATES):
+            return linux_constants.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]
+        if 0 <= family_idx < len(linux_constants.SOCK_FAMILY):
+            return linux_constants.SOCK_FAMILY[family_idx]
 
     def get_type(self):
-        return SOCK_TYPES.get(self.sk_type, "")
+        return linux_constants.SOCK_TYPES.get(self.sk_type, "")
 
     def get_inode(self):
         if not self.sk_socket:
@@ -1612,8 +1599,8 @@ def get_state(self):
         # 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]
+            if 0 <= state_idx < len(linux_constants.TCP_STATES):
+                return linux_constants.TCP_STATES[state_idx]
         else:
             # Return the generic socket state
             return self.sk.sk_socket.get_state()
@@ -1625,24 +1612,24 @@ def get_inode(self):
 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]
+        if 0 <= family_idx < len(linux_constants.SOCK_FAMILY):
+            return linux_constants.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)
+        protocol = linux_constants.IP_PROTOCOLS.get(self.sk.sk_protocol)
         if self.get_family() == "AF_INET6":
-            protocol = IPV6_PROTOCOLS.get(self.sk.sk_protocol, protocol)
+            protocol = linux_constants.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]
+            if 0 <= state_idx < len(linux_constants.TCP_STATES):
+                return linux_constants.TCP_STATES[state_idx]
         else:
             # Return the generic socket state
             return self.sk.sk_socket.get_state()
@@ -1725,8 +1712,8 @@ def get_dst_addr(self):
 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]
+        if 0 <= protocol_idx < len(linux_constants.NETLINK_PROTOCOLS):
+            return linux_constants.NETLINK_PROTOCOLS[protocol_idx]
 
     def get_state(self):
         # Return the generic socket state
@@ -1768,8 +1755,8 @@ def get_protocol(self):
         eth_proto = socket_module.htons(self.num)
         if eth_proto == 0:
             return None
-        elif eth_proto in ETH_PROTOCOLS:
-            return ETH_PROTOCOLS[eth_proto]
+        elif eth_proto in linux_constants.ETH_PROTOCOLS:
+            return linux_constants.ETH_PROTOCOLS[eth_proto]
         else:
             return f"0x{eth_proto:x}"
 
@@ -1781,13 +1768,13 @@ def get_state(self):
 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]
+        if 0 <= type_idx < len(linux_constants.BLUETOOTH_PROTOCOLS):
+            return linux_constants.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]
+        if 0 <= state_idx < len(linux_constants.BLUETOOTH_STATES):
+            return linux_constants.BLUETOOTH_STATES[state_idx]
 
 
 class xdp_sock(objects.StructType):
@@ -1905,7 +1892,7 @@ def get_last_cap_value(cls) -> int:
         Returns:
             int: The latest capability ID supported by the framework.
         """
-        return len(CAPABILITIES) - 1
+        return len(linux_constants.CAPABILITIES) - 1
 
     def get_kernel_cap_full(self) -> int:
         """Return the maximum value allowed for this kernel for a capability
@@ -1934,7 +1921,7 @@ def capabilities_to_string(cls, capabilities_bitfield: int) -> List[str]:
         """
 
         capabilities = []
-        for bit, name in enumerate(CAPABILITIES):
+        for bit, name in enumerate(linux_constants.CAPABILITIES):
             if capabilities_bitfield & (1 << bit) != 0:
                 capabilities.append(name)
 
@@ -1995,10 +1982,10 @@ def has_capability(self, capability: str) -> bool:
         Returns:
             bool: "True" if the given capability is enabled.
         """
-        if capability not in CAPABILITIES:
+        if capability not in linux_constants.CAPABILITIES:
             raise AttributeError(f"Unknown capability with name '{capability}'")
 
-        cap_value = 1 << CAPABILITIES.index(capability)
+        cap_value = 1 << linux_constants.CAPABILITIES.index(capability)
         return cap_value & self.get_capabilities() != 0