From af3b70320ecbbfc38c8ce51cc2aacf7ccd584580 Mon Sep 17 00:00:00 2001 From: Mike Auty Date: Fri, 6 Jan 2023 10:18:17 +0000 Subject: [PATCH] linux: Apply black linting to outstanding files --- .../framework/constants/linux/__init__.py | 20 +- volatility3/framework/plugins/linux/lsof.py | 27 +-- .../framework/plugins/linux/sockstat.py | 192 +++++++++++++----- .../framework/symbols/linux/__init__.py | 60 +++--- .../symbols/linux/extensions/__init__.py | 29 ++- 5 files changed, 219 insertions(+), 109 deletions(-) diff --git a/volatility3/framework/constants/linux/__init__.py b/volatility3/framework/constants/linux/__init__.py index 0c4d3c376..1b133eb42 100644 --- a/volatility3/framework/constants/linux/__init__.py +++ b/volatility3/framework/constants/linux/__init__.py @@ -13,7 +13,7 @@ PAGE_SHIFT = 12 """The value hard coded from the Linux Kernel (hence not extracted from the layer itself)""" # include/linux/sched.h -PF_KTHREAD = 0x00200000 # I'm a kernel thread +PF_KTHREAD = 0x00200000 # I'm a kernel thread # Standard well-defined IP protocols. # ref: include/uapi/linux/in.h @@ -139,13 +139,7 @@ SOCK_FAMILY = ( # Socket states # ref: include/uapi/linux/net.h -SOCKET_STATES = ( - "FREE", - "UNCONNECTED", - "CONNECTING", - "CONNECTED", - "DISCONNECTING" -) +SOCKET_STATES = ("FREE", "UNCONNECTED", "CONNECTING", "CONNECTED", "DISCONNECTING") # Netlink protocols # ref: include/uapi/linux/netlink.h @@ -188,17 +182,17 @@ ETH_PROTOCOLS = { 0x0007: "ETH_P_WAN_PPP", 0x0008: "ETH_P_PPP_MP", 0x0009: "ETH_P_LOCALTALK", - 0x000c: "ETH_P_CAN", - 0x000f: "ETH_P_CANFD", + 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", + 0x001A: "ETH_P_ARCNET", + 0x001B: "ETH_P_DSA", + 0x001C: "ETH_P_TRAILER", 0x0060: "ETH_P_LOOP", 0x00F6: "ETH_P_IEEE802154", 0x00F7: "ETH_P_CAIF", diff --git a/volatility3/framework/plugins/linux/lsof.py b/volatility3/framework/plugins/linux/lsof.py index 920aaf7f1..62bade1f1 100644 --- a/volatility3/framework/plugins/linux/lsof.py +++ b/volatility3/framework/plugins/linux/lsof.py @@ -46,10 +46,12 @@ class Lsof(plugins.PluginInterface): ] @classmethod - def list_fds(cls, - context: interfaces.context.ContextInterface, - symbol_table: str, - filter_func: Callable[[int], bool] = lambda _: False): + 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): @@ -62,18 +64,17 @@ class Lsof(plugins.PluginInterface): pid = int(task.pid) fd_generator = linux.LinuxUtilities.files_descriptors_for_process( - context, - linuxutils_symbol_table, - task) + 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) + 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 @@ -82,8 +83,8 @@ class Lsof(plugins.PluginInterface): yield (0, fields) def run(self): - pids = self.config.get('pid', None) - symbol_table = self.config['kernel'] + pids = self.config.get("pid", None) + symbol_table = self.config["kernel"] tree_grid_args = [("PID", int), ("Process", str), ("FD", int), ("Path", str)] - return renderers.TreeGrid(tree_grid_args, self._generator(pids, symbol_table)) \ No newline at end of file + return renderers.TreeGrid(tree_grid_args, self._generator(pids, symbol_table)) diff --git a/volatility3/framework/plugins/linux/sockstat.py b/volatility3/framework/plugins/linux/sockstat.py index ad3eee01f..f03a2ad8e 100644 --- a/volatility3/framework/plugins/linux/sockstat.py +++ b/volatility3/framework/plugins/linux/sockstat.py @@ -16,6 +16,7 @@ from volatility3.plugins.linux import lsof vollog = logging.getLogger(__name__) + class SockHandlers(interfaces.configuration.VersionableInterface): """Handles several socket families extracting the sockets information.""" @@ -56,7 +57,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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" + 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 @@ -64,7 +67,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]: + 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: @@ -86,7 +91,12 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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) + vollog.log( + constants.LOGLEVEL_V, + "Error processing socket family '%s': %s", + family, + e, + ) else: vollog.log(constants.LOGLEVEL_V, "Unsupported family '%s'", family) @@ -100,7 +110,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): return sock, sock_stat, socket_filter - def _update_socket_filters_info(self, sock: objects.Pointer, socket_filter: dict) -> None: + def _update_socket_filters_info( + self, sock: objects.Pointer, socket_filter: dict + ) -> None: """Get information from the socket and reuseport filters Args: @@ -117,7 +129,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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: + def _extract_socket_filter_info( + self, sock_filter: objects.Pointer, socket_filter: dict + ) -> None: """Get specific information for each type of filter Args: @@ -146,7 +160,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): if bpfprog_name: socket_filter["bpf_filter_name"] = bpfprog_name - def _unix_sock(self, sock: objects.StructType) -> Tuple[objects.StructType, Tuple[str, str, str]]: + def _unix_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_UNIX socket family Args: @@ -171,7 +187,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _inet_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_INET/6 socket families Args: @@ -191,7 +209,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _netlink_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_NETLINK socket family Args: @@ -221,7 +241,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _vsock_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_VSOCK socket family Args: @@ -241,7 +263,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _packet_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_PACKET socket family Args: @@ -262,7 +286,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _xdp_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_XDP socket family Args: @@ -304,7 +330,9 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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]]: + def _bluetooth_sock( + self, sock: objects.StructType + ) -> Tuple[objects.StructType, Tuple[str, str, str]]: """Handles the AF_BLUETOOTH socket family Args: @@ -324,11 +352,17 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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"): + 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") + 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") @@ -337,7 +371,10 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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") + 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") @@ -345,22 +382,31 @@ class SockHandlers(interfaces.configuration.VersionableInterface): 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") + 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") + 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) + 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.""" @@ -371,31 +417,48 @@ class Sockstat(plugins.PluginInterface): @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), + 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): + def list_sockets( + cls, + context: interfaces.context.ContextInterface, + symbol_table: str, + filter_func: Callable[[int], bool] = lambda _: False, + ): """Returns every single socket descriptor Args: @@ -433,7 +496,9 @@ class Sockstat(plugins.PluginInterface): if not d_inode: continue - socket_alloc = linux.LinuxUtilities.container_of(d_inode, "socket_alloc", "vfs_inode", vmlinux) + socket_alloc = linux.LinuxUtilities.container_of( + d_inode, "socket_alloc", "vfs_inode", vmlinux + ) socket = socket_alloc.socket if not (socket and socket.sk): @@ -466,7 +531,9 @@ class Sockstat(plugins.PluginInterface): Returns: `sock_stat` and `protocol` formatted. """ - sock_stat = [NotAvailableValue() if field is None else str(field) for field in sock_stat] + sock_stat = [ + NotAvailableValue() if field is None else str(field) for field in sock_stat + ] if protocol is None: protocol = NotAvailableValue() @@ -494,26 +561,49 @@ class Sockstat(plugins.PluginInterface): 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) + 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: + 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() + 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) + 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'] + pids = self.config.get("pids") + netns_id = self.config["netns"] + symbol_table = self.config["kernel"] tree_grid_args = [ ("NetNS", int), diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index 1d1419ae9..0d7cbb7e4 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -17,38 +17,38 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable): super().__init__(*args, **kwargs) # Set-up Linux specific types - self.set_type_class('file', extensions.struct_file) - self.set_type_class('list_head', extensions.list_head) - self.set_type_class('mm_struct', extensions.mm_struct) - self.set_type_class('super_block', extensions.super_block) - self.set_type_class('task_struct', extensions.task_struct) - self.set_type_class('vm_area_struct', extensions.vm_area_struct) - self.set_type_class('qstr', extensions.qstr) - 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('kobject', extensions.kobject) + self.set_type_class("file", extensions.struct_file) + self.set_type_class("list_head", extensions.list_head) + self.set_type_class("mm_struct", extensions.mm_struct) + self.set_type_class("super_block", extensions.super_block) + self.set_type_class("task_struct", extensions.task_struct) + self.set_type_class("vm_area_struct", extensions.vm_area_struct) + self.set_type_class("qstr", extensions.qstr) + 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("kobject", extensions.kobject) # Might not exist in the current symbols - self.optional_set_type_class('module', extensions.module) + self.optional_set_type_class("module", extensions.module) # Mount - self.set_type_class('vfsmount', extensions.vfsmount) + 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) + self.optional_set_type_class("mount", extensions.mount) + self.optional_set_type_class("mnt_namespace", extensions.mnt_namespace) # 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) + 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) + 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): @@ -322,7 +322,11 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): @classmethod def container_of( - cls, addr: int, type_name: str, member_name: str, vmlinux: interfaces.context.ModuleInterface + 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 @@ -343,4 +347,6 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): 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) + return vmlinux.object( + object_type=type_name, offset=container_addr, absolute=True + ) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 0ea61a3bb..55f139730 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -812,6 +812,7 @@ class vfsmount(objects.StructType): def get_mnt_root(self): return self.mnt_root + class kobject(objects.StructType): def reference_count(self): refcnt = self.kref.refcount @@ -842,6 +843,7 @@ 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"): @@ -851,12 +853,15 @@ class net(objects.StructType): 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)) + 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}") @@ -870,7 +875,9 @@ class socket(objects.StructType): except ValueError: return 0 - socket_alloc = linux.LinuxUtilities.container_of(self.vol.offset, "socket_alloc", "socket", kernel) + socket_alloc = linux.LinuxUtilities.container_of( + self.vol.offset, "socket_alloc", "socket", kernel + ) vfs_inode = socket_alloc.vfs_inode return vfs_inode.i_ino @@ -880,6 +887,7 @@ class socket(objects.StructType): 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 @@ -905,6 +913,7 @@ class sock(objects.StructType): return self.sk_socket.get_state() + class unix_sock(objects.StructType): def get_name(self): if not self.addr: @@ -932,6 +941,7 @@ class unix_sock(objects.StructType): 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 @@ -966,7 +976,7 @@ class inet_sock(objects.StructType): def get_dst_port(self): sk_common = self.sk.__sk_common if hasattr(sk_common, "skc_portpair"): - dport_le = sk_common.skc_portpair & 0xffff + dport_le = sk_common.skc_portpair & 0xFFFF elif hasattr(self, "dport"): dport_le = self.dport elif hasattr(self, "inet_dport"): @@ -999,7 +1009,9 @@ class inet_sock(objects.StructType): 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}") + vollog.debug( + f"Unable to read socket src address from {saddr.vol.offset:#x}" + ) return return socket_module.inet_ntop(family, addr_bytes) @@ -1028,11 +1040,14 @@ class inet_sock(objects.StructType): 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}") + 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 @@ -1043,6 +1058,7 @@ class netlink_sock(objects.StructType): # 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 @@ -1052,6 +1068,7 @@ class vsock_sock(objects.StructType): # 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) @@ -1066,6 +1083,7 @@ class packet_sock(objects.StructType): # 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 @@ -1077,6 +1095,7 @@ class bt_sock(objects.StructType): 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