From d532f3745438907aa654a5436bfcd04846763bf6 Mon Sep 17 00:00:00 2001 From: eve Date: Thu, 24 Apr 2025 09:37:39 +0100 Subject: [PATCH] Linux: sockstat contiuned work to breakdown large generator function --- .../framework/plugins/linux/sockscan.py | 128 +++++++++--------- 1 file changed, 65 insertions(+), 63 deletions(-) diff --git a/volatility3/framework/plugins/linux/sockscan.py b/volatility3/framework/plugins/linux/sockscan.py index 915a8e834..530b36c04 100644 --- a/volatility3/framework/plugins/linux/sockscan.py +++ b/volatility3/framework/plugins/linux/sockscan.py @@ -309,84 +309,86 @@ class Sockscan(plugins.PluginInterface): ) return None - def _generator(self, symbol_table_name: str): - """Scans for sockets. Each row represents a kernel socket. - - Args: - symbol_table_name: The name of the kernel module on which to operate - - Yields: - addr: Physical offset - 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) - """ - - # get vmlinux module from context in order to build objects and read symbols + def _build_sock_handler(self, symbol_table_name: str): vmlinux = self.context.modules[symbol_table_name] + init_task = vmlinux.object_from_symbol(symbol_name="init_task") + return sockstat.SockHandlers(self.context, symbol_table_name, init_task) - # get the memory layer that is to be scanned. + def _scan_for_sockets( + self, memory_layer, file_ops_needles, socket_destructor_needles + ): + return memory_layer.scan( + self.context, + scanners.MultiStringScanner(socket_destructor_needles | file_ops_needles), + self._progress_callback, + ) + + def _parse_scanner_result( + self, + match, + needle_addr, + file_ops_needles, + socket_destructor_needles, + symbol_table_name, + memory_layer_name, + f_op_offset, + sk_destruct_offset, + ): + + if match in socket_destructor_needles: + sock_physical_addr = needle_addr - sk_destruct_offset + psock = self._get_socket_from_sk_destruct( + sock_physical_addr, symbol_table_name, memory_layer_name + ) + elif match in file_ops_needles: + psock = self._walk_file_ops_needles( + symbol_table_name, memory_layer_name, needle_addr, f_op_offset + ) + sock_physical_addr = psock.vol.offset if psock else None + else: + psock = sock_physical_addr = None + + return psock, sock_physical_addr + + def _get_socket_from_sk_destruct( + self, sock_physical_addr, symbol_table_name, memory_layer_name + ): + vmlinux = self.context.modules[symbol_table_name] + return self.context.object( + vmlinux.symbol_table_name + constants.BANG + "sock", + offset=sock_physical_addr, + layer_name=memory_layer_name, + native_layer_name=vmlinux.layer_name, + ) + + def _generator(self, symbol_table_name: str): memory_layer_name = self._find_memory_layer_name(symbol_table_name) memory_layer = self.context.layers[memory_layer_name] + sock_handler = self._build_sock_handler(symbol_table_name) - # use the init process to build a sock handler - # TODO: look into options so that sockstat.SockHandlers so that process_sock can - # be used without a task object. - init_task = vmlinux.object_from_symbol(symbol_name="init_task") - sock_handler = sockstat.SockHandlers(self.context, symbol_table_name, init_task) - - # get progress_callback in order to use this in the scanners. - # TODO: perhaps add more detail to progress, showing method in progress and number of hits found - progress_callback = self._progress_callback - - # Method 1 - find sockets by file operations, then follow pointers to sockets file_ops_needles, f_op_offset = self._find_file_ops_needles(symbol_table_name) - - # Method 2 - find sockets by socket destructor directly inside sock objects socket_destructor_needles, sk_destruct_offset = self._find_sk_destruct_needles( symbol_table_name ) - # TODO Method 3 - find sock by sk_error_report symbols - # sk_error_report_symbol_names = ['sock_def_error_report', 'inet_sk_rebuild_header', 'inet_listen'] - # this would be similar to Method 2, but using a different pointer within sock. - - # add a set of seen addresses to stop possible duplication of results. seen_sock_physical_addr = set() - # Using the calculated needles, scan the memory layer and attempt to parse the sockets located. - for needle_addr, match in memory_layer.scan( - self.context, - scanners.MultiStringScanner(socket_destructor_needles | file_ops_needles), - progress_callback, + for needle_addr, match in self._scan_for_sockets( + memory_layer, file_ops_needles, socket_destructor_needles ): - psock = None - sock_physical_addr = None + psock, sock_physical_addr = self._parse_scanner_result( + match, + needle_addr, + file_ops_needles, + socket_destructor_needles, + symbol_table_name, + memory_layer_name, + f_op_offset, + sk_destruct_offset, + ) - # if match is from socket_destructor_needles simply calculate the offset to the sock - if match in socket_destructor_needles: - sock_physical_addr = needle_addr - sk_destruct_offset - psock = self.context.object( - vmlinux.symbol_table_name + constants.BANG + "sock", - offset=sock_physical_addr, - layer_name=memory_layer_name, - native_layer_name=vmlinux.layer_name, - ) - - # if match is from file_ops_needles attempt to walk from file object to the sock - if match in file_ops_needles: - psock = self._walk_file_ops_needles( - symbol_table_name, memory_layer_name, needle_addr, f_op_offset - ) - - if psock is not None and sock_physical_addr not in seen_sock_physical_addr: + if psock and sock_physical_addr not in seen_sock_physical_addr: seen_sock_physical_addr.add(sock_physical_addr) - fields = self._extract_sock_fields(psock, sock_handler) if fields: yield (0, fields)