mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-13 21:27:39 +02:00
Linux: update sockscan to scan memory layer only once for needles
This commit is contained in:
Eve
@@ -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):
|
||||
|
||||
|
||||
Reference in New Issue
Block a user