linux_utilities show deleted fd + lsof files_only arg

Author: SolitudePy

volatility3/framework/plugins/linux/lsof.py

@@ -137,6 +137,12 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
                 element_type=int,
                 optional=True,
             ),
+            requirements.BooleanRequirement(
+                name="files_only",
+                description="Include only file descriptors of type file",
+                optional=True,
+                default=False,
+            ),
         ]
 
     @classmethod
@@ -145,6 +151,7 @@ def list_fds(
         context: interfaces.context.ContextInterface,
         vmlinux_module_name: str,
         filter_func: Callable[[int], bool] = lambda _: False,
+        include_files_only: bool = False,
     ) -> Iterable[FDInternal]:
         """Enumerates open file descriptors in tasks
 
@@ -167,16 +174,20 @@ def list_fds(
                 linuxutils_symbol_table = task.vol.type_name.split(constants.BANG)[0]
 
             fd_generator = linux.LinuxUtilities.files_descriptors_for_process(
-                context, linuxutils_symbol_table, task
+                context, linuxutils_symbol_table, task, files_only=include_files_only
             )
 
             for fd_fields in fd_generator:
                 yield FDInternal(task=task, fd_fields=fd_fields)
 
     def _generator(self, pids, vmlinux_module_name):
         filter_func = pslist.PsList.create_pid_filter(pids)
+        include_files_only = self.config.get("files_only")
         for fd_internal in self.list_fds(
-            self.context, vmlinux_module_name, filter_func=filter_func
+            self.context,
+            vmlinux_module_name,
+            filter_func=filter_func,
+            include_files_only=include_files_only,
         ):
             fd_user = fd_internal.to_user()
             yield (0, dataclasses.astuple(fd_user))

volatility3/framework/symbols/linux/__init__.py

@@ -101,6 +101,7 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
 
     _version = (2, 3, 1)
     _required_framework_version = (2, 0, 0)
+    deleted = " (deleted)"
 
     framework.require_interface_version(*_required_framework_version)
 
@@ -168,6 +169,7 @@ def do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> Union[None, str]:
             # vfsmnt can be the vfsmount object itself (>=3.3) or a vfsmount * (<3.3)
             return ""
 
+        inode = dentry.d_inode
         path_reversed = []
         smeared = False
         while (
@@ -191,6 +193,7 @@ def do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> Union[None, str]:
 
             parent = dentry.d_parent
             dname = dentry.d_name.name_as_str()
+
             # empty dentry names are most likely
             # the result of smearing
             if not dname:
@@ -204,6 +207,9 @@ def do_get_path(cls, rdentry, rmnt, dentry, vfsmnt) -> Union[None, str]:
             # path would be /foo/bar/baz, but bar is missing due to smear the results
             # returned here will show /foo//baz. Note the // for the missing dname.
             return f"<potentially smeared> {path}"
+        print(path, inode.i_nlink)
+        if inode and inode.is_readable() and inode.is_valid() and inode.i_nlink == 0:
+            path += LinuxUtilities.deleted
         return path
 
     @classmethod
@@ -260,7 +266,7 @@ def _get_new_sock_pipe_path(cls, context, task, filp) -> str:
                     pre_name = name.dereference().cast(
                         "string", max_length=255, errors="replace"
                     )
-                    return "/" + pre_name + " (deleted)"
+                    return "/" + pre_name + LinuxUtilities.deleted
                 else:
                     pre_name = ""
 
@@ -301,7 +307,7 @@ def _get_new_sock_pipe_path(cls, context, task, filp) -> str:
         return f"{pre_name}:[{inode.i_ino:d}]"
 
     @classmethod
-    def path_for_file(cls, context, task, filp) -> str:
+    def path_for_file(cls, context, task, filp, files_only) -> str:
         """Returns a file (or sock pipe) pathname relative to the task's root directory.
 
         A 'file' structure doesn't have enough information to properly restore its
@@ -340,7 +346,7 @@ def path_for_file(cls, context, task, filp) -> str:
         except exceptions.InvalidAddressException:
             dname_is_valid = False
 
-        if dname_is_valid:
+        if dname_is_valid and not files_only:
             ret = LinuxUtilities._get_new_sock_pipe_path(context, task, filp)
         else:
             ret = LinuxUtilities._get_path_file(task, filp)
@@ -353,6 +359,7 @@ def files_descriptors_for_process(
         context: interfaces.context.ContextInterface,
         symbol_table: str,
         task: interfaces.objects.ObjectInterface,
+        files_only: bool = False,
     ):
         try:
             files = task.files
@@ -376,7 +383,9 @@ def files_descriptors_for_process(
 
         for fd_num, filp in enumerate(fds):
             if filp and filp.is_readable():
-                full_path = LinuxUtilities.path_for_file(context, task, filp)
+                full_path = LinuxUtilities.path_for_file(
+                    context, task, filp, files_only
+                )
 
                 yield fd_num, filp, full_path