Linux: add basic versions of linux.ip.Link and linux.ip.Addr plugins

volatility3/framework/constants/linux/__init__.py

@@ -281,3 +281,39 @@
 )
 
 ELF_MAX_EXTRACTION_SIZE = 1024 * 1024 * 1024 * 4 - 1
+
+# net_device_flags was not always an enum, so these hard coded values
+# are needed when processing older kernels
+# include/uapi/linux/if.h
+NET_DEVICE_FLAGS = {
+    "IFF_UP": 0x1,
+    "IFF_BROADCAST": 0x2,
+    "IFF_DEBUG": 0x4,
+    "IFF_LOOPBACK": 0x8,
+    "IFF_POINTOPOINT": 0x10,
+    "IFF_NOTRAILERS": 0x20,
+    "IFF_RUNNING": 0x40,
+    "IFF_NOARP": 0x80,
+    "IFF_PROMISC": 0x100,
+    "IFF_ALLMULTI": 0x200,
+    "IFF_MASTER": 0x400,
+    "IFF_SLAVE": 0x800,
+    "IFF_MULTICAST": 0x1000,
+    "IFF_PORTSEL": 0x2000,
+    "IFF_AUTOMEDIA": 0x4000,
+    "IFF_DYNAMIC": 0x8000,
+    "IFF_LOWER_UP": 0x10000,
+    "IFF_DORMANT": 0x20000,
+    "IFF_ECHO": 0x40000,
+}
+
+# ref include/uapi/linux/if.h
+RFC2863OPERATIONALSTATE = (
+    "UNKNOWN",
+    "NOTPRESENT",
+    "DOWN",
+    "LOWERLAYERDOWN",
+    "TESTING",
+    "DORMANT",
+    "UP",
+)

volatility3/framework/plugins/linux/ip.py

@@ -0,0 +1,167 @@
+# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+"""A module containing a collection of plugins that produce data typically
+produced from the `ip` command, e.g. ip addr, ip link, ip neigh, ip route etc."""
+
+import logging
+from volatility3.framework import renderers, constants, interfaces
+from volatility3.framework.configuration import requirements
+from volatility3.framework.interfaces import plugins
+from volatility3.framework.renderers import NotAvailableValue
+
+
+vollog = logging.getLogger(__name__)
+
+
+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):
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            )
+        ]
+
+    def _get_net_device_link_info(self, net_device):
+        device_name = net_device.get_device_name()
+        hardware_address = net_device.get_hardware_address()
+        operational_state = net_device.get_operational_state()
+        mtu = net_device.mtu
+        qdisc_name = net_device.get_qdisc_name()
+        net_device_flags_str = ",".join(net_device.get_flag_names())
+
+        yield (
+            device_name,
+            hardware_address,
+            operational_state,
+            mtu,
+            qdisc_name,
+            net_device_flags_str,
+        )
+
+    @classmethod
+    def list_net_devices(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> (interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface):
+        """Lists all the network devices in the primary layer.
+
+        Args:
+            context: The context to retrieve required elements (layers, symbol tables) from
+            vmlinux_module_name: The name of the kernel module on which to operate
+
+        Yields:
+            Tuple of objects (net, net_device)
+        """
+        vmlinux = context.modules[vmlinux_module_name]
+
+        # find net_devices using aviable symbols
+        if vmlinux.has_symbol("net_namespace_list"):
+            vollog.debug(f"found net_namespace_list symbol")
+            table_name = vmlinux.symbol_table_name
+            net_symbol_name = table_name + constants.BANG + "net"
+            net_device_symbol_name = table_name + constants.BANG + "net_device"
+            nethead = vmlinux.object_from_symbol(symbol_name="net_namespace_list")
+            for net in nethead.to_list(net_symbol_name, "list"):
+                for net_device in net.dev_base_head.to_list(
+                    net_device_symbol_name, "dev_list"
+                ):
+                    yield (net, net_device)
+
+        elif vmlinux.has_symbol("dev_base"):
+            # TODO: add support for old kernels. <2.6.24 did not have net_namespace_list
+            raise ("Not yet implimented")
+
+        else:
+            raise TypeError(
+                "This plugin requires the either the net_namespace_list or dev_base symbol. This ",
+                "symbol is not present in the supplied symbol table. This means you are either ",
+                "analyzing an unsupported kernel version or that your symbol table is corrupt.",
+            )
+
+    def _generator(self):
+        vmlinux_module_name = self.config["kernel"]
+
+        for net, net_device in self.list_net_devices(self.context, vmlinux_module_name):
+            for device_link_info in self._get_net_device_link_info(net_device):
+                try:
+                    net_ns = net.get_inode()
+                except AttributeError:
+                    net_ns = NotAvailableValue()
+                result = (net_ns,) + device_link_info
+                yield (0, result)
+
+    def run(self):
+        return renderers.TreeGrid(
+            [
+                ("NS", int),
+                ("Interface", str),
+                ("MAC", str),
+                ("State", str),
+                ("MTU", int),
+                ("Qdisc", str),
+                ("Flags", str),
+            ],
+            self._generator(),
+        )
+
+
+class Addr(plugins.PluginInterface):
+    """Lists information about network interfaces similar to `ip addr show`"""
+
+    _required_framework_version = (2, 0, 0)
+
+    @classmethod
+    def get_requirements(cls):
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.PluginRequirement(name="Link", plugin=Link, version=(1, 0, 0)),
+        ]
+
+    def _get_net_device_addrs(self, net_device):
+        device_name = net_device.get_device_name()
+        operational_state = net_device.get_operational_state()
+        promiscuous = net_device.get_promiscuous_state()
+        # TODO: also include ifalias in net_device ?
+        for ip_addr in net_device.get_ip_addresses():
+            yield (
+                device_name,
+                ip_addr,
+                operational_state,
+                promiscuous,
+            )
+
+    def _generator(self, nets):
+        for net, net_device in nets:
+            for device_addr_info in self._get_net_device_addrs(net_device):
+                try:
+                    net_ns = net.get_inode()
+                except AttributeError:
+                    net_ns = NotAvailableValue()
+                result = (net_ns,) + device_addr_info
+                yield (0, result)
+
+    def run(self):
+        return renderers.TreeGrid(
+            [
+                ("NS", int),
+                ("Interface", str),
+                ("IP", str),
+                ("State", str),
+                ("Promiscuous", bool),
+            ],
+            self._generator(Link.list_net_devices(self.context, self.config["kernel"])),
+        )

volatility3/framework/symbols/linux/__init__.py

@@ -46,6 +46,7 @@ def __init__(self, *args, **kwargs) -> None:
         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_device", extensions.net_device)
         # 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)

volatility3/framework/symbols/linux/extensions/__init__.py

@@ -19,6 +19,12 @@
 from volatility3.framework.objects import utility
 from volatility3.framework.symbols import generic, linux, intermed
 from volatility3.framework.symbols.linux.extensions import elf
+from volatility3.framework.renderers import conversion
+from volatility3.framework.symbols.wrappers import Flags
+from volatility3.framework.constants.linux import (
+    NET_DEVICE_FLAGS,
+    RFC2863OPERATIONALSTATE,
+)
 
 vollog = logging.getLogger(__name__)
 
@@ -1581,3 +1587,77 @@ def has_capability(self, capability: str) -> bool:
 
         cap_value = 1 << CAPABILITIES.index(capability)
         return cap_value & self.get_capabilities() != 0
+
+
+class net_device(objects.StructType):
+    def get_hardware_address(self):
+        parent_layer = self._context.layers[self.vol.layer_name]
+        if self.has_member("perm_addr"):
+            raw_addr = self.perm_addr[0 : self.addr_len]
+        else:  # perm_addr is not found in older kernels
+            raw_addr = parent_layer.read(self.dev_addr, self.addr_len, pad=True)
+        addr = ":".join([f"{byte:02x}" for byte in raw_addr])
+        return addr
+
+    def get_operational_state(self):
+        return RFC2863OPERATIONALSTATE[self.operstate]
+
+    def get_promiscuous_state(self):
+        return True if self.flags & NET_DEVICE_FLAGS["IFF_PROMISC"] else False
+
+    def get_device_name(self):
+        return utility.array_to_string(self.name)
+
+    def get_qdisc_name(self):
+        return utility.array_to_string(self.qdisc.ops.id)
+
+    def get_flag_names(self) -> List[str]:
+        """Return the net_deivce flags as a list of strings"""
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        try:
+            # read flags using the net_device_flags enum
+            net_device_flags_enum = vmlinux.get_enumeration("net_device_flags")
+            choices = net_device_flags_enum.choices
+
+        except exceptions.SymbolError:
+            vollog.debug(
+                f"Unable to find net_device_flags enum in ISF, using hard coded enum for flag names"
+            )
+            choices = NET_DEVICE_FLAGS
+
+        net_device_flags_enum_flags = Flags(choices=choices)
+        flags = net_device_flags_enum_flags(self.flags)
+
+        # format flags to string, drop IFF_ to match `ip link` output
+        flags = [flag.replace("IFF_", "") for flag in flags]
+        return flags
+
+    def get_ip_addresses(self) -> List[str]:
+        return self.get_ipv4_addresses() + self.get_ipv6_addresses()
+
+    def get_ipv4_addresses(self) -> List[str]:
+        ipv4_addresses = []
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+
+        # follow chain to all in_ifaddr
+        in_ifaddr_ptr = self.ip_ptr.ifa_list
+        while in_ifaddr_ptr > 0:
+            in_ifaddr = vmlinux.object("in_ifaddr", in_ifaddr_ptr, absolute=True)
+            ip_addr = conversion.convert_ipv4(in_ifaddr.ifa_address)
+            ip_prefix = in_ifaddr.ifa_prefixlen
+            ipv4_addresses.append(f"{ip_addr}/{ip_prefix}")
+            # process next in list
+            in_ifaddr_ptr = in_ifaddr.ifa_next
+        return ipv4_addresses
+
+    def get_ipv6_addresses(self) -> List[str]:
+        ipv6_addresses = []
+        symbol_table_name = self.get_symbol_table_name()
+        inet6_ifaddr_symbol_name = symbol_table_name + constants.BANG + "inet6_ifaddr"
+        for inet6_ifaddr in self.ip6_ptr.addr_list.to_list(
+            inet6_ifaddr_symbol_name, "if_list"
+        ):
+            ip_addr = conversion.convert_ipv6(inet6_ifaddr.addr.in6_u.u6_addr32)
+            ip_prefix = inet6_ifaddr.prefix_len
+            ipv6_addresses.append(f"{ip_addr}/{ip_prefix}")
+        return ipv6_addresses