From 67291ce90fcbea2462293082b1f5d246dfca1b24 Mon Sep 17 00:00:00 2001 From: Eve Date: Wed, 27 Mar 2024 09:29:01 +0000 Subject: [PATCH] Linux: update sockscan to scan memory layer only once for needles --- .../framework/plugins/linux/sockscan.py | 388 ++++++++++-------- 1 file changed, 210 insertions(+), 178 deletions(-) diff --git a/volatility3/framework/plugins/linux/sockscan.py b/volatility3/framework/plugins/linux/sockscan.py index 52b33d9c1..d02757238 100644 --- a/volatility3/framework/plugins/linux/sockscan.py +++ b/volatility3/framework/plugins/linux/sockscan.py @@ -4,13 +4,12 @@ import logging import struct -from typing import Callable, Tuple, List, Dict +from typing import List, Set -from volatility3.framework import interfaces, exceptions, constants, objects +from volatility3.framework import exceptions, constants 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 sockstat from volatility3.framework import symbols @@ -41,6 +40,62 @@ class Sockscan(plugins.PluginInterface): ), ] + def _canonicalize_symbol_addrs( + self, symbol_table_name: List[str], symbol_names: str + ) -> Set[bytes]: + """Takes a list of symbol names and converts the address of each to the bytes + as they would appear in memory so that they can be scanned for. + + Symbols that cannot be found are ignored and not included in the results. + + Args: + symbol_table_name: The name of the kernel module on which to operate + symbol_names: A list of symbol names to be looked up + + Returns: + A set of bytes which are the packed addresses. + """ + # get vmlinux module from context in order to build objects and read symbols + vmlinux = self.context.modules[symbol_table_name] + + # get kernel layer from context so that it's dependencies can be found, and therefore scanned. + # kernel layer will be virtual and built ontop of a physical layer. + kernel_layer = self.context.layers[vmlinux.layer_name] + + # detmine if kernel is 64bit or not. The plugin scans for pointers and these need to formated + # to the correct size so that they can be accurately located in the physical layer. + if symbols.symbol_table_is_64bit(self.context, vmlinux.symbol_table_name): + pack_format = "Q" # 64 bit + else: + pack_format = "I" # 32 bit + + packed_needles = set() + for symbol_name in symbol_names: + try: + needle_addr = vmlinux.object_from_symbol(symbol_name).vol.offset + except exceptions.SymbolError: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Unable to find symbol {symbol_name} this will not be scanned for.", + ) + continue + # use canonicalize to set the appropriate sign extension for the addr + addr = kernel_layer.canonicalize(needle_addr) + packed_addr = struct.pack(pack_format, addr) + packed_needles.add(packed_addr) + vollog.log( + constants.LOGLEVEL_VVVV, + f"Will scan for {symbol_name} using the bytes: {packed_addr.hex()}", + ) + + # make a warning if no symbols at all could be resolved. + if len(packed_needles) == 0: + vollog.warning( + f"_canonicalize_symbol_addrs was unable to resolve any symbols, use -vvvv for more information." + ) + + return packed_needles + def _generator(self, symbol_table_name: str): """Scans for sockets. Each row represents a kernel socket. @@ -48,6 +103,7 @@ class Sockscan(plugins.PluginInterface): 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) @@ -65,15 +121,9 @@ class Sockscan(plugins.PluginInterface): # kernel layer will be virtual and built ontop of a physical layer. kernel_layer = self.context.layers[vmlinux.layer_name] - # detmine if kernel is 64bit or not. The plugin scans for pointers and these need to formated - # to the correct size so that they can be accurately located in the physical layer. - if symbols.symbol_table_is_64bit(self.context, vmlinux.symbol_table_name): - pack_format = "Q" # 64 bit - else: - pack_format = "I" # 32 bit - # TODO: Update plugin to support multiple dependencies. e.g. a memory layer and swap file. # This is a shared problem with psscan and having a generic solution would be useful. + # Find the memory layer to scan, and provide warnings if more than one is located. if len(kernel_layer.dependencies) > 1: vollog.warning( @@ -95,91 +145,23 @@ class Sockscan(plugins.PluginInterface): init_task = vmlinux.object_from_symbol(symbol_name="init_task") sock_handler = sockstat.SockHandlers(vmlinux, init_task) - # set to track seen sockets so that results are not duplicated between methods - sock_physical_addresses = set() - # 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 - # TODO: update scanning logic so that all needles can be scanned for at the same time - # this would allow the results to be shown as the scanning is happening and would - # make the plugin faster. It would require working out which needle caused the match - # and applying the logic at that point to get to the socket. - # Method 1 - find sockets by file operations, then follow pointers to sockets - file_ops_symbol_names = ["socket_file_ops", "sockfs_dentry_operations"] - file_ops_needles = [] - for symbol_name in file_ops_symbol_names: - - # TODO: handle cases where symbol is not found - needle_addr = vmlinux.object_from_symbol(symbol_name).vol.offset - # use canonicalize to set the appropriate sign extension for the addr - addr = kernel_layer.canonicalize(needle_addr) - packed_addr = struct.pack(pack_format, addr) - file_ops_needles.append(packed_addr) - vollog.log( - constants.LOGLEVEL_VVVV, - f"Method 1 will scan for {symbol_name} using the bytes: {packed_addr.hex()}", - ) - + file_ops_symbol_names = [ + "socket_file_ops", + "sockfs_dentry_operations", + ] + file_ops_needles = self._canonicalize_symbol_addrs( + symbol_table_name, file_ops_symbol_names + ) # get file struct to find the offset to the f_op pointer # this is so that the file object can be created at the correct offset, # the results of the scanner will be for the f_op member within the file f_op_offset = vmlinux.get_type("file").members["f_op"][0] - for addr, _ in memory_layer.scan( - self.context, - scanners.MultiStringScanner(file_ops_needles), - progress_callback, - ): - try: - # create file in the memory_layer, the native layer matches the - # kernel so that pointers can be followed - pfile = self.context.object( - vmlinux.symbol_table_name + constants.BANG + "file", - offset=addr - f_op_offset, - layer_name=memory_layer_name, - native_layer_name=vmlinux.layer_name, - ) - dentry = pfile.get_dentry() - if not dentry: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Skipping file at {hex(addr)} as unable to locate dentry", - ) - continue - - d_inode = dentry.d_inode - if not d_inode: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Skipping file at {hex(addr)} as unable to locate inode for dentry", - ) - continue - - socket_alloc = linux.LinuxUtilities.container_of( - d_inode, "socket_alloc", "vfs_inode", vmlinux - ) - socket = socket_alloc.socket - if not (socket and socket.sk): - vollog.log( - constants.LOGLEVEL_VVVV, - f"Skipping file at {hex(addr)} as socket created by LinuxUtilities.container_of is invalid", - ) - continue - - # sucessfully trversed from file to sock, this will exist in the - # kernel layer, and need to be translated to the memory layer. - sock = socket.sk.dereference() - sock_physical_addresses.add(kernel_layer.translate(sock.vol.offset)[0]) - - except exceptions.InvalidAddressException as error: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Unable to follow file at {hex(addr)} to socket due to invalid address: {error}", - ) - # Method 2 - find sockets by socket destructor directly inside sock objects socket_destructor_symbol_names = [ "sock_def_destruct", @@ -188,114 +170,164 @@ class Sockscan(plugins.PluginInterface): "netlink_sock_destruct", "inet_sock_destruct", ] - - socket_destructor_needles = [] - for socket_destructor_symbol_name in socket_destructor_symbol_names: - addr = kernel_layer.canonicalize( - vmlinux.get_symbol(socket_destructor_symbol_name).address - + vmlinux.offset - ) - packed_addr = struct.pack(pack_format, addr) - socket_destructor_needles.append(packed_addr) - vollog.log( - constants.LOGLEVEL_VVVV, - f"Method 2 will scan for {socket_destructor_symbol_name} using the bytes: {packed_addr.hex()}", - ) - + socket_destructor_needles = self._canonicalize_symbol_addrs( + symbol_table_name, socket_destructor_symbol_names + ) # get sock struct to find the offset to the sk_destruct pointer # this is so that the sock object can be created at the correct offset, # the results of the scanner will be for the sk_destruct member within the scock sk_destruct_offset = vmlinux.get_type("sock").members["sk_destruct"][0] - for addr, _ in memory_layer.scan( - self.context, - scanners.MultiStringScanner(socket_destructor_needles), - progress_callback, - ): - sock_physical_addresses.add(addr - sk_destruct_offset) - # 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. - # now that the set of results has been created, process them and display the results - for addr in sorted(sock_physical_addresses): - psock = self.context.object( - vmlinux.symbol_table_name + constants.BANG + "sock", - offset=addr, - layer_name=memory_layer_name, - native_layer_name=vmlinux.layer_name, - ) - try: - sock_type = psock.get_type() + # 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, + ): + psock = None + sock_physical_addr = None - family = psock.get_family() - # remove results with no family - if family == None: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Skipping socket at {hex(addr)} as unable to determin family.", - ) - continue - - # TODO: invesitgate options for more invalid address handling in proccess_sock - # and the later formatting on it's results. - sock_fields = sock_handler.process_sock(psock) - # if no sock_fields we're able to be extracted then skip this result. - if not sock_fields: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Skipping socket at {hex(addr)} as unable to process with SockHandlers.", - ) - continue - - sock, sock_stat, extended = sock_fields - src, src_port, dst, dst_port, state = sock_stat - protocol = sock.get_protocol() - - # format results - src = NotAvailableValue() if src is None else str(src) - src_port = NotAvailableValue() if src_port is None else str(src_port) - dst = NotAvailableValue() if dst is None else str(dst) - dst_port = NotAvailableValue() if dst_port is None else str(dst_port) - state = NotAvailableValue() if state is None else str(state) - protocol = NotAvailableValue() if protocol is None else str(protocol) - # extended attributes is a dict, so this is formated to string show each - # key and value pair, seperated with a comma. - socket_filter_str = ( - ",".join(f"{k}={v}" for k, v in extended.items()) - if extended - else NotAvailableValue() + # 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, ) - # remove empty results - if (src == "0.0.0.0" or isinstance(src, NotAvailableValue)) and ( - dst == "0.0.0.0" or isinstance(src, NotAvailableValue) - ): - if state == "UNCONNECTED": - continue - elif src_port == "0" and dst_port == "0": + # if match is from file_ops_needles attempt to walk from file object to + # the sock + if match in file_ops_needles: + try: + # create file in the memory_layer, the native layer matches the + # kernel so that pointers can be followed + sock_physical_addr = needle_addr - f_op_offset + pfile = self.context.object( + vmlinux.symbol_table_name + constants.BANG + "file", + offset=sock_physical_addr, + layer_name=memory_layer_name, + native_layer_name=vmlinux.layer_name, + ) + dentry = pfile.get_dentry() + if not dentry: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Skipping file at {hex(needle_addr)} as unable to locate dentry", + ) continue - fields = ( - format_hints.Hex(sock.vol.offset), - family, - sock_type, - protocol, - src, - src_port, - dst, - dst_port, - state, - socket_filter_str, - ) + d_inode = dentry.d_inode + if not d_inode: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Skipping file at {hex(needle_addr)} as unable to locate inode for dentry", + ) + continue - yield (0, fields) - except exceptions.InvalidAddressException as error: - vollog.log( - constants.LOGLEVEL_VVVV, - f"Unable create results for socket at {hex(addr)} to invalid address: {error}", - ) + socket_alloc = linux.LinuxUtilities.container_of( + d_inode, "socket_alloc", "vfs_inode", vmlinux + ) + socket = socket_alloc.socket + if not (socket and socket.sk): + vollog.log( + constants.LOGLEVEL_VVVV, + f"Skipping file at {hex(needle_addr)} as socket created by LinuxUtilities.container_of is invalid", + ) + continue + + # sucessfully trversed from file to sock, this will exist in the + # kernel layer, and need to be translated to the memory layer. + psock = socket.sk.dereference() + except exceptions.InvalidAddressException as error: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Unable to follow file at {hex(needle_addr)} to socket due to invalid address: {error}", + ) + + if psock is not None: + try: + sock_type = psock.get_type() + + family = psock.get_family() + # remove results with no family + if family == None: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Skipping socket at {hex(sock_physical_addr)} as unable to determin family.", + ) + continue + + # TODO: invesitgate options for more invalid address handling in proccess_sock + # and the later formatting of it's results. + sock_fields = sock_handler.process_sock(psock) + # if no sock_fields we're able to be extracted then skip this result. + if not sock_fields: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Skipping socket at {hex(sock_physical_addr)} as unable to process with SockHandlers.", + ) + continue + + sock, sock_stat, extended = sock_fields + src, src_port, dst, dst_port, state = sock_stat + protocol = sock.get_protocol() + + # format results + src = NotAvailableValue() if src is None else str(src) + src_port = ( + NotAvailableValue() if src_port is None else str(src_port) + ) + dst = NotAvailableValue() if dst is None else str(dst) + dst_port = ( + NotAvailableValue() if dst_port is None else str(dst_port) + ) + state = NotAvailableValue() if state is None else str(state) + protocol = ( + NotAvailableValue() if protocol is None else str(protocol) + ) + # extended attributes is a dict, so this is formated to string show each + # key and value pair, seperated with a comma. + socket_filter_str = ( + ",".join(f"{k}={v}" for k, v in extended.items()) + if extended + else NotAvailableValue() + ) + + # remove empty results + if (src == "0.0.0.0" or isinstance(src, NotAvailableValue)) and ( + dst == "0.0.0.0" or isinstance(src, NotAvailableValue) + ): + if state == "UNCONNECTED": + continue + elif src_port == "0" and dst_port == "0": + continue + + fields = ( + format_hints.Hex(sock_physical_addr), + family, + sock_type, + protocol, + src, + src_port, + dst, + dst_port, + state, + socket_filter_str, + ) + + yield (0, fields) + except exceptions.InvalidAddressException as error: + vollog.log( + constants.LOGLEVEL_VVVV, + f"Unable create results for socket at {hex(sock_physical_addr)} due to invalid address: {error}", + ) def run(self):