WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-08-30 11:49:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

linux: Apply black linting to outstanding files

Browse Source
This commit is contained in:
Mike Auty
2023-01-06 10:18:17 +00:00
parent a9928d66f1
commit af3b70320e
5 changed files with 219 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/constants/linux/__init__.py
+7 -13
View File
@@ -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",
volatility3/framework/plugins/linux/lsof.py
+14 -13
View File
@@ -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))
return renderers.TreeGrid(tree_grid_args, self._generator(pids, symbol_table))
volatility3/framework/plugins/linux/sockstat.py
+141 -51
View File
@@ -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),
volatility3/framework/symbols/linux/__init__.py
+33 -27
View File
@@ -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
)
volatility3/framework/symbols/linux/extensions/__init__.py
+24 -5
View File
@@ -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