diff --git a/volatility3/framework/__init__.py b/volatility3/framework/__init__.py index 2c9444130..176eb2242 100644 --- a/volatility3/framework/__init__.py +++ b/volatility3/framework/__init__.py @@ -51,7 +51,7 @@ def require_interface_version(*args) -> None: if args[1] > interface_version()[1]: raise RuntimeError( "Framework interface version {} is an older revision than the required version {}".format( - ".".join([str(x) for x in interface_version()[0:1]]), ".".join([str(x) for x in args[0:2]]))) + ".".join([str(x) for x in interface_version()[0:2]]), ".".join([str(x) for x in args[0:2]]))) class NonInheritable(object): diff --git a/volatility3/framework/constants/__init__.py b/volatility3/framework/constants/__init__.py index 5060906d5..472a743e6 100644 --- a/volatility3/framework/constants/__init__.py +++ b/volatility3/framework/constants/__init__.py @@ -39,7 +39,7 @@ BANG = "!" # We use the SemVer 2.0.0 versioning scheme VERSION_MAJOR = 2 # Number of releases of the library with a breaking change -VERSION_MINOR = 1 # Number of changes that only add to the interface +VERSION_MINOR = 2 # Number of changes that only add to the interface VERSION_PATCH = 0 # Number of changes that do not change the interface VERSION_SUFFIX = "" diff --git a/volatility3/framework/plugins/linux/mountinfo.py b/volatility3/framework/plugins/linux/mountinfo.py new file mode 100644 index 000000000..6f3cb712d --- /dev/null +++ b/volatility3/framework/plugins/linux/mountinfo.py @@ -0,0 +1,220 @@ +# This file is Copyright 2021 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +# Author: Gustavo Moreira + +import logging +from collections import namedtuple +from typing import Tuple, List, Iterable, Union + +from volatility3.framework import renderers, interfaces +from volatility3.framework.configuration import requirements +from volatility3.framework.interfaces import plugins +from volatility3.plugins.linux import pslist + +vollog = logging.getLogger(__name__) + +MountInfoData = namedtuple("MountInfoData", ("mnt_id", "parent_id", "st_dev", "mnt_root_path", "path_root", + "mnt_opts", "fields", "mnt_type", "devname", "sb_opts")) + +class MountInfo(plugins.PluginInterface): + """Lists mount points on processes mount namespaces""" + + _required_framework_version = (2, 2, 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"]), + requirements.PluginRequirement(name="pslist", + plugin=pslist.PsList, version=(2, 0, 0)), + requirements.ListRequirement(name="pids", + description="Filter on specific process IDs.", + element_type=int, + optional=True), + requirements.ListRequirement(name="mntns", + description="Filter results by mount namespace. " + "Otherwise, all of them are shown.", + element_type=int, + optional=True), + requirements.BooleanRequirement(name="mount-format", + description="Shows a brief summary of the mount points information " + "with similar output format to the older /proc/[pid]/mounts or the " + "user-land command 'mount -l'.", + optional=True, + default=False), + ] + + @classmethod + def _do_get_path(cls, mnt, fs_root) -> Union[None, str]: + """It mimics the Linux kernel prepend_path function.""" + vfsmnt = mnt.mnt + dentry = vfsmnt.get_mnt_root() + + path_reversed = [] + while dentry != fs_root.dentry or vfsmnt.vol.offset != fs_root.mnt: + if dentry == vfsmnt.get_mnt_root() or dentry.is_root(): + parent = mnt.get_mnt_parent().dereference() + # Escaped? + if dentry != vfsmnt.get_mnt_root(): + return None + + # Global root? + if mnt.vol.offset != parent.vol.offset: + dentry = mnt.get_mnt_mountpoint() + mnt = parent + vfsmnt = mnt.mnt + continue + + return None + + parent = dentry.d_parent + dname = dentry.d_name.name_as_str() + path_reversed.append(dname.strip("/")) + dentry = parent + + path = "/" + "/".join(reversed(path_reversed)) + return path + + @classmethod + def get_mountinfo(cls, mnt, task) -> Union[None, Tuple[int, int, str, str, str, List[str], + List[str], str, str, List[str]]]: + """Extract various information about a mount point. + It mimics the Linux kernel show_mountinfo function. + """ + mnt_root = mnt.get_mnt_root() + if not mnt_root: + return None + + path_root = cls._do_get_path(mnt, task.fs.root) + if path_root is None: + return None + + mnt_root_path = mnt_root.path() + superblock = mnt.get_mnt_sb() + + mnt_id: int = mnt.mnt_id + parent_id: int = mnt.mnt_parent.mnt_id + + st_dev = f"{superblock.major}:{superblock.minor}" + + mnt_opts: List[str] = [] + mnt_opts.append(mnt.get_flags_access()) + mnt_opts.extend(mnt.get_flags_opts()) + + # Tagged fields + fields: List[str] = [] + if mnt.is_shared(): + fields.append(f"shared:{mnt.mnt_group_id}") + + if mnt.is_slave(): + master = mnt.mnt_master.mnt_group_id + fields.append(f"master:{master}") + dominating_id = mnt.get_dominating_id(task.fs.root) + if dominating_id and dominating_id != master: + fields.append(f"propagate_from:{dominating_id}") + + if mnt.is_unbindable(): + fields.append("unbindable") + + mnt_type = superblock.get_type() + + devname = mnt.get_devname() + if not devname: + devname = "none" + + sb_opts: List[str] = [] + sb_opts.append(superblock.get_flags_access()) + sb_opts.extend(superblock.get_flags_opts()) + + return MountInfoData(mnt_id, parent_id, st_dev, mnt_root_path, path_root, mnt_opts, fields, + mnt_type, devname, sb_opts) + + def _get_tasks_mountpoints(self, tasks: Iterable[interfaces.objects.ObjectInterface], per_namespace: bool): + seen_namespaces = set() + for task in tasks: + if not (task and task.fs and task.fs.root and task.nsproxy and task.nsproxy.mnt_ns): + # This task doesn't have all the information required + continue + + mnt_namespace = task.nsproxy.mnt_ns + mnt_ns_id = mnt_namespace.get_inode() + + if per_namespace: + if mnt_ns_id in seen_namespaces: + continue + else: + seen_namespaces.add(mnt_ns_id) + + for mount in mnt_namespace.get_mount_points(): + yield task, mount, mnt_ns_id + + def _generator( + self, + tasks: Iterable[interfaces.objects.ObjectInterface], + mnt_ns_ids: List[int], + mount_format: bool, + per_namespace: bool) -> Iterable[Tuple[int, Tuple]]: + + for task, mnt, mnt_ns_id in self._get_tasks_mountpoints(tasks, per_namespace): + if mnt_ns_ids and mnt_ns_id not in mnt_ns_ids: + continue + + mnt_info = self.get_mountinfo(mnt, task) + if mnt_info is None: + continue + + if mount_format: + all_opts = set() + all_opts.update(mnt_info.mnt_opts) + all_opts.update(mnt_info.sb_opts) + all_opts_str = ",".join(all_opts) + + extra_fields_values = [mnt_info.devname, mnt_info.path_root, mnt_info.mnt_type, all_opts_str] + else: + mnt_opts_str = ",".join(mnt_info.mnt_opts) + fields_str = " ".join(mnt_info.fields) + sb_opts_str = ",".join(mnt_info.sb_opts) + + extra_fields_values = [mnt_info.mnt_id, mnt_info.parent_id, mnt_info.st_dev, mnt_info.mnt_root_path, + mnt_info.path_root, mnt_opts_str, fields_str, mnt_info.mnt_type, + mnt_info.devname, sb_opts_str] + + fields_values = [mnt_ns_id] + if not per_namespace: + fields_values.append(task.pid) + fields_values.extend(extra_fields_values) + + yield (0, fields_values) + + def run(self): + pids = self.config.get('pids') + mount_ns_ids = self.config.get('mntns') + mount_format = self.config.get('mount-format') + + pid_filter = pslist.PsList.create_pid_filter(pids) + tasks = pslist.PsList.list_tasks(self.context, self.config['kernel'], filter_func=pid_filter) + + columns = [("MNT_NS_ID", int)] + # The PID column does not make sense when a PID filter is not specified. In that case, the default behavior is + # to displays the mountpoints per namespace. + if pids: + columns.append(("PID", int)) + per_namespace = False + else: + per_namespace = True + + if self.config.get('mount-format'): + extra_columns = [("DEVNAME", str), ("PATH", str), ("FSTYPE", str), ("MNT_OPTS", str)] + else: + # /proc/[pid]/mountinfo output format + extra_columns = [("MOUNT ID", int), ("PARENT_ID", int), ("MAJOR:MINOR", str), ("ROOT", str), + ("MOUNT_POINT", str), ("MOUNT_OPTIONS", str), ("FIELDS", str), ("FSTYPE", str), + ("MOUNT_SRC", str), ("SB_OPTIONS", str)] + + columns.extend(extra_columns) + + return renderers.TreeGrid(columns, self._generator(tasks, mount_ns_ids, mount_format, per_namespace)) diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index 8d7f00e06..0c5ce395c 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -29,6 +29,7 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable): self.set_type_class('files_struct', extensions.files_struct) self.set_type_class('vfsmount', extensions.vfsmount) self.set_type_class('kobject', extensions.kobject) + self.set_type_class('mnt_namespace', extensions.mnt_namespace) if 'module' in self.types: self.set_type_class('module', extensions.module) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 2217317f4..6792ab19c 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -288,6 +288,29 @@ class super_block(objects.StructType): # include/linux/kdev_t.h MINORBITS = 20 + # Superblock flags + SB_RDONLY = 1 # Mount read-only + SB_NOSUID = 2 # Ignore suid and sgid bits + SB_NODEV = 4 # Disallow access to device special files + SB_NOEXEC = 8 # Disallow program execution + SB_SYNCHRONOUS = 16 # Writes are synced at once + SB_MANDLOCK = 64 # Allow mandatory locks on an FS + SB_DIRSYNC = 128 # Directory modifications are synchronous + SB_NOATIME = 1024 # Do not update access times + SB_NODIRATIME = 2048 # Do not update directory access times + SB_SILENT = 32768 + SB_POSIXACL = (1 << 16) # VFS does not apply the umask + SB_KERNMOUNT = (1 << 22) # this is a kern_mount call + SB_I_VERSION = (1 << 23) # Update inode I_version field + SB_LAZYTIME = (1 << 25) # Update the on-disk [acm]times lazily + + SB_OPTS = { + SB_SYNCHRONOUS: "sync", + SB_DIRSYNC: "dirsync", + SB_MANDLOCK: "mand", + SB_LAZYTIME: "lazytime" + } + @property def major(self) -> int: return self.s_dev >> self.MINORBITS @@ -296,6 +319,20 @@ class super_block(objects.StructType): def minor(self) -> int: return self.s_dev & ((1 << self.MINORBITS) - 1) + def get_flags_access(self) -> str: + return 'ro' if self.s_flags & self.SB_RDONLY else 'rw' + + def get_flags_opts(self) -> Iterable[str]: + sb_opts = [self.SB_OPTS[sb_opt] for sb_opt in self.SB_OPTS if sb_opt & self.s_flags] + return sb_opts + + def get_type(self): + mnt_sb_type = utility.pointer_to_string(self.s_type.name, count=255) + if self.s_subtype: + mnt_sb_subtype = utility.pointer_to_string(self.s_subtype, count=255) + mnt_sb_type += "." + mnt_sb_subtype + return mnt_sb_type + class vm_area_struct(objects.StructType): perm_flags = { @@ -415,7 +452,50 @@ class qstr(objects.StructType): class dentry(objects.StructType): def path(self) -> str: - return self.d_name.name_as_str() + """Based on __dentry_path Linux kernel function""" + reversed_path = [] + dentry_seen = set() + current_dentry = self + while (not current_dentry.is_root() and + current_dentry.vol.offset not in dentry_seen): + parent = current_dentry.d_parent + reversed_path.append(current_dentry.d_name.name_as_str()) + dentry_seen.add(current_dentry.vol.offset) + current_dentry = parent + return "/" + "/".join(reversed(reversed_path)) + + def is_root(self) -> bool: + return self.vol.offset == self.d_parent + + def is_subdir(self, old_dentry): + """Is this dentry a subdirectory of old_dentry? + + Returns true if this dentry is a subdirectory of the parent (at any depth). + Otherwise, it returns false. + """ + if self.vol.offset == old_dentry: + return True + + return self.d_ancestor(old_dentry) + + def d_ancestor(self, ancestor_dentry): + """Search for an ancestor + + Returns the ancestor dentry which is a child of "ancestor_dentry", + if "ancestor_dentry" is an ancestor of "child_dentry", else None. + """ + + dentry_seen = set() + current_dentry = self + while (not current_dentry.is_root() and + current_dentry.vol.offset not in dentry_seen): + if current_dentry.d_parent == ancestor_dentry.vol.offset: + return current_dentry + + dentry_seen.add(current_dentry.vol.offset) + current_dentry = current_dentry.d_parent + + return None class struct_file(objects.StructType): @@ -510,6 +590,27 @@ class files_struct(objects.StructType): class mount(objects.StructType): + MNT_NOSUID = 0x01 + MNT_NODEV = 0x02 + MNT_NOEXEC = 0x04 + MNT_NOATIME = 0x08 + MNT_NODIRATIME = 0x10 + MNT_RELATIME = 0x20 + MNT_READONLY = 0x40 + MNT_SHRINKABLE = 0x100 + MNT_WRITE_HOLD = 0x200 + MNT_SHARED = 0x1000 + MNT_UNBINDABLE = 0x2000 + + MNT_FLAGS = { + MNT_NOSUID: "nosuid", + MNT_NODEV: "nodev", + MNT_NOEXEC: "noexec", + MNT_NOATIME: "noatime", + MNT_NODIRATIME: "nodiratime", + MNT_RELATIME: "relatime", + } + def get_mnt_sb(self): if self.has_member("mnt"): return self.mnt.mnt_sb @@ -540,6 +641,82 @@ class mount(objects.StructType): def get_mnt_mountpoint(self): return self.mnt_mountpoint + def get_flags_access(self) -> str: + return "ro" if self.get_mnt_flags() & self.MNT_READONLY else "rw" + + def get_flags_opts(self) -> Iterable[str]: + flags = [self.MNT_FLAGS[mntflag] for mntflag in self.MNT_FLAGS if mntflag & self.get_mnt_flags()] + return flags + + def is_shared(self) -> bool: + return self.get_mnt_flags() & self.MNT_SHARED + + def is_unbindable(self) -> bool: + return self.get_mnt_flags() & self.MNT_UNBINDABLE + + def is_slave(self) -> bool: + return self.mnt_master and self.mnt_master.vol.offset != 0 + + def get_devname(self) -> str: + return utility.pointer_to_string(self.mnt_devname, count=255) + + def has_parent(self) -> bool: + return self.vol.offset != self.mnt_parent + + def get_dominating_id(self, root) -> int: + """Get ID of closest dominating peer group having a representative under the given root.""" + mnt_seen = set() + current_mnt = self.mnt_master + while (current_mnt and + current_mnt.vol.offset != 0 and + current_mnt.vol.offset not in mnt_seen): + peer = current_mnt.get_peer_under_root(self.mnt_ns, root) + if peer and peer.vol.offset != 0: + return peer.mnt_group_id + + mnt_seen.add(current_mnt.vol.offset) + current_mnt = current_mnt.mnt_master + return 0 + + def get_peer_under_root(self, ns, root): + """Return true if path is reachable from root. + It mimics the kernel function is_path_reachable(), ref: fs/namespace.c + """ + mnt_seen = set() + current_mnt = self + while current_mnt.vol.offset not in mnt_seen: + if current_mnt.mnt_ns == ns and current_mnt.is_path_reachable(current_mnt.mnt.mnt_root, root): + return current_mnt + + mnt_seen.add(current_mnt.vol.offset) + current_mnt = current_mnt.next_peer() + if current_mnt.vol.offset == self.vol.offset: + break + + return None + + def is_path_reachable(self, current_dentry, root): + """Return true if path is reachable. + It mimics the kernel function with same name, ref fs/namespace.c: + """ + mnt_seen = set() + current_mnt = self + while (current_mnt.mnt.vol.offset != root.mnt and + current_mnt.has_parent() and + current_mnt.vol.offset not in mnt_seen): + + current_dentry = current_mnt.mnt_mountpoint + mnt_seen.add(current_mnt.vol.offset) + current_mnt = current_mnt.mnt_parent + + return current_mnt.mnt.vol.offset == root.mnt and current_dentry.is_subdir(root.dentry) + + def next_peer(self): + table_name = self.vol.type_name.split(constants.BANG)[0] + mount_struct = "{0}{1}mount".format(table_name, constants.BANG) + offset = self._context.symbol_space.get_type(mount_struct).relative_child_offset("mnt_share") + + return self._context.object(mount_struct, self.vol.layer_name, offset=self.mnt_share.next.vol.offset - offset) class vfsmount(objects.StructType): @@ -581,3 +758,22 @@ class kobject(objects.StructType): ret = refcnt.refs.counter return ret + +class mnt_namespace(objects.StructType): + def get_inode(self): + if self.has_member("proc_inum"): + return self.proc_inum + elif self.ns.has_member("inum"): + return self.ns.inum + else: + raise AttributeError("Unable to find mnt_namespace inode") + + def get_mount_points(self): + table_name = self.vol.type_name.split(constants.BANG)[0] + mnt_type = table_name + constants.BANG + "mount" + if not self._context.symbol_space.has_type(mnt_type): + # Old kernels ~ 2.6 + mnt_type = table_name + constants.BANG + "vfsmount" + + for mount in self.list.to_list(mnt_type, "mnt_list"): + yield mount