Linux: Add the linux.ip.Link plugin by @eve-mem

- On top of the @eve-mem, I've added the queue length field to mimic the ip link command.
- Furthermore, I've included some functions to export the network device flags exactly as they are presented to userland
This commit is contained in:
Gustavo Moreira
2024-02-02 20:57:26 +11:00
parent 260fbd8d24
commit 7b1c75cc87
2 changed files with 208 additions and 4 deletions
+89 -2
View File
@@ -6,7 +6,6 @@ from typing import List
from volatility3.framework import interfaces, renderers, constants
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
class Addr(plugins.PluginInterface):
@@ -30,7 +29,7 @@ class Addr(plugins.PluginInterface):
mac_addr = net_dev.get_mac_address()
promisc = net_dev.promisc
operational_state = net_dev.get_operational_state()
iface_name = utility.array_to_string(net_dev.name)
iface_name = net_dev.get_device_name()
iface_ifindex = net_dev.ifindex
try:
net_ns_id = net_dev.get_net_namespace_id()
@@ -80,3 +79,91 @@ class Addr(plugins.PluginInterface):
]
return renderers.TreeGrid(headers, self._generator())
class Link(plugins.PluginInterface):
"""Lists information about network interfaces similar to `ip link show`"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
)
]
def _gather_net_dev_link_info(self, net_device):
mac_addr = net_device.get_mac_address()
operational_state = net_device.get_operational_state()
iface_name = net_device.get_device_name()
mtu = net_device.mtu
qdisc_name = net_device.get_qdisc_name()
qlen = net_device.get_queue_length()
# Format flags to string. Drop IFF_ to match iproute2 'ip link' output.
# Also, note that iproute2 removes IFF_RUNNING, see print_link_flags()
flags_list = [
flag.replace("IFF_", "")
for flag in net_device.get_flag_names()
if flag != "IFF_RUNNING"
]
flags_str = ",".join(flags_list)
yield iface_name, mac_addr, operational_state, mtu, qdisc_name, qlen, flags_str
@classmethod
def list_net_devices(
cls,
vmlinux: interfaces.context.ModuleInterface,
) -> (interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface):
"""Lists network devices
Args:
vmlinux (ModuleInterface): The kernel symbols object
Yields:
tuple:
net: Network namespace
net_device: Network device structure
"""
table_name = vmlinux.symbol_table_name
net_type_symname = table_name + constants.BANG + "net"
net_device_symname = table_name + constants.BANG + "net_device"
# 'net_namespace_list' exists from kernels >= 2.6.24
net_namespace_list = vmlinux.object_from_symbol("net_namespace_list")
for net in net_namespace_list.to_list(net_type_symname, "list"):
for net_device in net.dev_base_head.to_list(net_device_symname, "dev_list"):
yield net, net_device
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
for net_dev, net_device in self.list_net_devices(vmlinux):
for device_link_info in self._gather_net_dev_link_info(net_device):
try:
net_ns_id = net_dev.get_net_namespace_id()
except AttributeError:
net_ns_id = renderers.NotAvailableValue()
fields = [net_ns_id, *device_link_info]
yield (0, fields)
def run(self):
headers = [
("NS", int),
("Interface", str),
("MAC", str),
("State", str),
("MTU", int),
("Qdisc", str),
("Qlen", int),
("Flags", str),
]
return renderers.TreeGrid(headers, self._generator())
@@ -21,6 +21,7 @@ from volatility3.framework import exceptions, objects, interfaces, symbols
from volatility3.framework.layers import linear
from volatility3.framework.objects import utility
from volatility3.framework.symbols import generic, linux, intermed
from volatility3.framework.symbols.wrappers import Flags
from volatility3.framework.symbols.linux.extensions import elf
vollog = logging.getLogger(__name__)
@@ -1237,6 +1238,14 @@ class net(objects.StructType):
class net_device(objects.StructType):
def get_device_name(self) -> str:
"""Return the network device name
Returns:
str: The network device name
"""
return utility.array_to_string(self.name)
def _format_as_mac_address(self, hwaddr):
return ":".join([f"{x:02x}" for x in hwaddr[: self.addr_len]])
@@ -1281,13 +1290,105 @@ class net_device(objects.StructType):
"""Return the net_device flag value based on the flag name"""
return self._get_flag_choices().get(name, UnparsableValue())
def _get_netdev_state_t(self):
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
try:
# At least from kernels 2.6.30
return vmlinux.get_enumeration("netdev_state_t")
except exceptions.SymbolError:
raise exceptions.VolatilityException(
"Unsupported kernel or wrong ISF. Cannot find 'netdev_state_t' enumeration"
)
def is_running(self) -> bool:
"""Test if the network device has been brought up
Based on netif_running()
Returns:
bool: True if the device is UP
"""
netdev_state_t_enum = self._get_netdev_state_t()
# It should be safe. netdev_state_t::__LINK_STATE_START has been available since
# at least kernels 2.6.30
return (
self.state & (1 << netdev_state_t_enum.choices["__LINK_STATE_START"]) != 0
)
def is_carrier_ok(self) -> bool:
"""Check if carrier is present on network device
Based on netif_carrier_ok()
Returns:
bool: True if carrier present
"""
netdev_state_t_enum = self._get_netdev_state_t()
# It should be safe. netdev_state_t::__LINK_STATE_NOCARRIER has been available
# since at least kernels 2.6.30
return (
self.state & (1 << netdev_state_t_enum.choices["__LINK_STATE_NOCARRIER"])
== 0
)
def is_dormant(self) -> bool:
"""Check if the network device is dormant
Based on netif_dormant(()
Returns:
bool: True if the network device is dormant
"""
netdev_state_t_enum = self._get_netdev_state_t()
# It should be safe. netdev_state_t::__LINK_STATE_DORMANT has been available
# since at least kernels 2.6.30
return (
self.state & (1 << netdev_state_t_enum.choices["__LINK_STATE_DORMANT"]) != 0
)
def is_operational(self) -> bool:
"""Test if the carrier is operational
Based on netif_oper_up()
Returns:
bool: True if the device is UP
"""
return self.get_operational_state() in ("UP", "UNKNOWN")
def get_flag_names(self) -> List[str]:
"""Return the net_device flags as a list of strings
"""Return the net_device flags as a list of strings.
This is the combination of flags exported through kernel APIs to userspace.
Based on dev_get_flags()
Returns:
List[str]: A list of flag names
"""
return list(self._get_flag_choices())
choices = self._get_flag_choices()
clear_flags = choices.get("IFF_PROMISC", 0)
clear_flags |= choices.get("IFF_ALLMULTI", 0)
clear_flags |= choices.get("IFF_RUNNING", 0)
clear_flags |= choices.get("IFF_LOWER_UP", 0)
clear_flags |= choices.get("IFF_DORMANT", 0)
clear_gflags = choices.get("IFF_PROMISC", 0)
clear_gflags |= choices.get("IFF_ALLMULTI)", 0)
flags = (self.flags & ~clear_flags) | (self.gflags & ~clear_gflags)
if self.is_running():
if self.is_operational():
flags |= choices.get("IFF_RUNNING", 0)
if self.is_carrier_ok():
flags |= choices.get("IFF_LOWER_UP", 0)
if self.is_dormant():
flags |= choices.get("IFF_DORMANT", 0)
net_device_flags_enum_flags = Flags(choices)
net_device_flags = net_device_flags_enum_flags(flags)
# It's preferable to provide a deterministic list of items. i.e. for testing
return sorted(net_device_flags)
@property
def promisc(self):
@@ -1327,6 +1428,22 @@ class net_device(objects.StructType):
vollog.warning(f"Invalid net_device operational state '{self.operstate}'")
return UnparsableValue()
def get_qdisc_name(self) -> str:
"""Return the network device queuing discipline (qdisc) name
Returns:
str: A string with the queuing discipline (qdisc) name
"""
return utility.array_to_string(self.qdisc.ops.id)
def get_queue_length(self) -> int:
"""Return the netwrok device transmision qeueue length (qlen)
Returns:
int: the netwrok device transmision qeueue length (qlen)
"""
return self.tx_queue_len
class in_device(objects.StructType):
def get_addresses(self):