Merge pull request #593 from gcmoreira/mountinfo

Mountinfo

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):

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 = ""
 

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))

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)

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