WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-11 04:07:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Linux: sockstat contiuned work to breakdown large generator function

Browse Source
This commit is contained in:
eve
2025-04-24 09:37:39 +01:00
parent 704f0f023d
commit d532f37454
1 changed files with 65 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/plugins/linux/sockscan.py
+65 -63
View File
@@ -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)