From 90b327e63253404a98cc5a79e3cecaa1b773048c Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Sat, 24 Aug 2024 11:45:29 +1000 Subject: [PATCH] PR review fixes: Make mountinfo.get_superblocks() a classmethod and adapt the code using it. --- .../framework/plugins/linux/mountinfo.py | 19 +++++++++--- .../framework/plugins/linux/pagecache.py | 31 ++++++++++++------- 2 files changed, 33 insertions(+), 17 deletions(-) diff --git a/volatility3/framework/plugins/linux/mountinfo.py b/volatility3/framework/plugins/linux/mountinfo.py index dfb2e2f52..1eaec77bf 100644 --- a/volatility3/framework/plugins/linux/mountinfo.py +++ b/volatility3/framework/plugins/linux/mountinfo.py @@ -37,7 +37,7 @@ class MountInfo(plugins.PluginInterface): _required_framework_version = (2, 2, 0) - _version = (1, 1, 0) + _version = (1, 2, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -143,8 +143,8 @@ class MountInfo(plugins.PluginInterface): sb_opts, ) + @staticmethod def _get_tasks_mountpoints( - self, tasks: Iterable[interfaces.objects.ObjectInterface], filtered_by_pids: bool = False, ): @@ -247,17 +247,26 @@ class MountInfo(plugins.PluginInterface): "Could not filter by mount namespace id. This field is not available in this kernel." ) - def get_superblocks(self): + @classmethod + def get_superblocks( + cls, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> Iterable[interfaces.objects.ObjectInterface]: """Yield file system superblocks based on the task's mounted filesystems. + 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: super_block: Kernel's struct super_block object """ # No filter so that we get all the mount namespaces from all tasks - tasks = pslist.PsList.list_tasks(self.context, self.config["kernel"]) + tasks = pslist.PsList.list_tasks(context, vmlinux_module_name) seen_sb_ptr = set() - for task, mnt, _mnt_ns_id in self._get_tasks_mountpoints(tasks): + for task, mnt, _mnt_ns_id in cls._get_tasks_mountpoints(tasks): path_root = linux.LinuxUtilities.get_path_mnt(task, mnt) if not path_root: continue diff --git a/volatility3/framework/plugins/linux/pagecache.py b/volatility3/framework/plugins/linux/pagecache.py index e384cbabf..e54891480 100644 --- a/volatility3/framework/plugins/linux/pagecache.py +++ b/volatility3/framework/plugins/linux/pagecache.py @@ -115,7 +115,7 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface): architectures=["Intel32", "Intel64"], ), requirements.PluginRequirement( - name="mountinfo", plugin=mountinfo.MountInfo, version=(1, 1, 0) + name="mountinfo", plugin=mountinfo.MountInfo, version=(1, 2, 0) ), requirements.ListRequirement( name="type", @@ -204,22 +204,22 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface): def get_inodes( cls, context: interfaces.context.ContextInterface, - config_path: str, + vmlinux_module_name: str, ) -> Iterable[InodeInternal]: """Retrieves the inodes from the superblocks Args: context: The context that the plugin will operate within - config_path: The path to configuration data within the context configuration data + vmlinux_module_name: The name of the kernel module on which to operate Yields: An InodeInternal object """ - superblocks_iter = mountinfo.MountInfo( + superblocks_iter = mountinfo.MountInfo.get_superblocks( context=context, - config_path=config_path, - ).get_superblocks() + vmlinux_module_name=vmlinux_module_name, + ) seen_inodes = set() seen_dentries = set() @@ -289,11 +289,13 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface): yield inode_in def _generator(self): - vmlinux = self.context.modules[self.config["kernel"]] + vmlinux_module_name = self.config["kernel"] + vmlinux = self.context.modules[vmlinux_module_name] vmlinux_layer = self.context.layers[vmlinux.layer_name] inodes_iter = self.get_inodes( - context=self.context, config_path=self.config_path + context=self.context, + vmlinux_module_name=vmlinux_module_name, ) types_filter = self.config["type"] @@ -316,12 +318,15 @@ class Files(plugins.PluginInterface, timeliner.TimeLinerInterface): These need not be generated in any particular order, sorting will be done later """ - vmlinux = self.context.modules[self.config["kernel"]] + vmlinux_module_name = self.config["kernel"] + vmlinux = self.context.modules[vmlinux_module_name] vmlinux_layer = self.context.layers[vmlinux.layer_name] inodes_iter = self.get_inodes( - context=self.context, config_path=self.config_path + context=self.context, + vmlinux_module_name=vmlinux_module_name, ) + for inode_in in inodes_iter: inode_out = inode_in.to_user(vmlinux_layer) description = f"Cached Inode for {inode_out.path}" @@ -450,7 +455,8 @@ class InodePages(plugins.PluginInterface): vollog.error("Unable to write to file (%s): %s", filename, e) def _generator(self): - vmlinux = self.context.modules[self.config["kernel"]] + vmlinux_module_name = self.config["kernel"] + vmlinux = self.context.modules[vmlinux_module_name] vmlinux_layer = self.context.layers[vmlinux.layer_name] if self.config["inode"] and self.config["find"]: @@ -459,7 +465,8 @@ class InodePages(plugins.PluginInterface): if self.config["find"]: inodes_iter = Files.get_inodes( - context=self.context, config_path=self.config_path + context=self.context, + vmlinux_module_name=vmlinux_module_name, ) for inode_in in inodes_iter: if inode_in.path == self.config["find"]: