Merge pull request #1613 from Abyss-W4tcher/linux_pagecache_recoverfs

New plugin: linux.pagecache.recoverfs

Author: ikelos

volatility3/framework/constants/_version.py

@@ -1,7 +1,7 @@
 # We use the SemVer 2.0.0 versioning scheme
 VERSION_MAJOR = 2  # Number of releases of the library with a breaking change
-VERSION_MINOR = 20  # Number of changes that only add to the interface
-VERSION_PATCH = 1  # Number of changes that do not change the interface
+VERSION_MINOR = 21  # Number of changes that only add to the interface
+VERSION_PATCH = 0  # Number of changes that do not change the interface
 VERSION_SUFFIX = ""
 
 PACKAGE_VERSION = (

volatility3/framework/plugins/linux/pagecache.py

@@ -5,11 +5,15 @@
 import math
 import logging
 import datetime
+import time
+import tarfile
 from dataclasses import dataclass, astuple
 from typing import IO, List, Set, Type, Iterable, Tuple
+from io import BytesIO
+from pathlib import PurePath
 
 from volatility3.framework.constants import architectures
-from volatility3.framework import renderers, interfaces, exceptions
+from volatility3.framework import constants, renderers, interfaces, exceptions
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.interfaces import plugins
 from volatility3.framework.configuration import requirements
@@ -625,3 +629,260 @@ def run(self):
         return renderers.TreeGrid(
             headers, Files.format_fields_with_headers(headers, self._generator())
         )
+
+
+class RecoverFs(plugins.PluginInterface):
+    """Recovers the cached filesystem (directories, files, symlinks) into a compressed tarball.
+
+    Details: level 0 directories are named after the UUID of the parent superblock; metadata aren't replicated to extracted objects; objects modification time is set to the plugin run time; absolute symlinks
+    are converted to relative symlinks to prevent referencing the analyst's filesystem.
+    Troubleshooting: to fix extraction errors related to long paths, please consider using https://github.com/mxmlnkn/ratarmount.
+    """
+
+    _version = (1, 0, 0)
+    _required_framework_version = (2, 21, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Linux kernel",
+                architectures=architectures.LINUX_ARCHS,
+            ),
+            requirements.PluginRequirement(
+                name="files", plugin=Files, version=(1, 1, 0)
+            ),
+            requirements.PluginRequirement(
+                name="inodepages", plugin=InodePages, version=(3, 0, 0)
+            ),
+            requirements.ChoiceRequirement(
+                name="compression_format",
+                description="Compression format (default: gz)",
+                choices=["gz", "bz2", "xz"],
+                default="gz",
+                optional=True,
+            ),
+        ]
+
+    def _tar_add_reg_inode(
+        self,
+        context: interfaces.context.ContextInterface,
+        layer_name: str,
+        tar: tarfile.TarFile,
+        reg_inode_in: InodeInternal,
+        path_prefix: str = "",
+        mtime: float = None,
+    ) -> int:
+        """Extracts a REG inode content and writes it to a TarFile object.
+
+        Args:
+            context: The context on which to operate
+            layer_name: The name of the layer on which to operate
+            tar: The TarFile object to write to
+            reg_inode_in: The inode to extract content from
+            path_prefix: A custom path prefix to prepend the inode path with
+            mtime: The modification time to set the TarInfo object to
+
+        Returns:
+            The number of extracted bytes
+        """
+        inode_content_buffer = BytesIO()
+        InodePages.write_inode_content_to_stream(
+            context, layer_name, reg_inode_in.inode, inode_content_buffer
+        )
+        inode_content_buffer.seek(0)
+        handle_buffer_size = inode_content_buffer.getbuffer().nbytes
+
+        tar_info = tarfile.TarInfo(path_prefix + reg_inode_in.path)
+        # The tarfile module only has read support for sparse files:
+        # https://docs.python.org/3.12/library/tarfile.html#tarfile.LNKTYPE:~:text=and%20longlink%20extensions%2C-,read%2Donly%20support,-for%20all%20variants
+        tar_info.type = tarfile.REGTYPE
+        tar_info.size = handle_buffer_size
+        tar_info.mode = 0o444
+        if mtime is not None:
+            tar_info.mtime = mtime
+        tar.addfile(tar_info, inode_content_buffer)
+
+        return handle_buffer_size
+
+    def _tar_add_dir(
+        self,
+        tar: tarfile.TarFile,
+        directory_path: str,
+        mtime: float = None,
+    ) -> None:
+        """Adds a directory path to a TarFile object, based on a DIR inode.
+
+        Args:
+            tar: The TarFile object to write to
+            directory_path: The directory path to create
+            mtime: The modification time to set the TarInfo object to
+        """
+        tar_info = tarfile.TarInfo(directory_path)
+        tar_info.type = tarfile.DIRTYPE
+        tar_info.mode = 0o755
+        if mtime is not None:
+            tar_info.mtime = mtime
+        tar.addfile(tar_info)
+
+    def _tar_add_lnk(
+        self,
+        tar: tarfile.TarFile,
+        symlink_source: str,
+        symlink_dest: str,
+        symlink_source_prefix: str = "",
+        mtime: float = None,
+    ) -> None:
+        """Adds a symlink to a TarFile object.
+
+        Args:
+            tar: The TarFile object to write to
+            symlink_source: The symlink source path
+            symlink_dest: The symlink target/destination
+            symlink_source_prefix: A custom path prefix to prepend the symlink source with
+            mtime: The modification time to set the TarInfo object to
+        """
+        # Patch symlinks pointing to absolute paths,
+        # to prevent referencing the host filesystem.
+        if symlink_dest.startswith("/"):
+            relative_dest = PurePath(symlink_dest).relative_to(PurePath("/"))
+            # Remove the leading "/" to prevent an extra undesired "../" in the output
+            symlink_dest = (
+                PurePath(
+                    *[".."] * len(PurePath(symlink_source.lstrip("/")).parent.parts)
+                )
+                / relative_dest
+            ).as_posix()
+        tar_info = tarfile.TarInfo(symlink_source_prefix + symlink_source)
+        tar_info.type = tarfile.SYMTYPE
+        tar_info.linkname = symlink_dest
+        tar_info.mode = 0o444
+        if mtime is not None:
+            tar_info.mtime = mtime
+        tar.addfile(tar_info)
+
+    def _generator(self):
+        vmlinux_module_name = self.config["kernel"]
+        vmlinux = self.context.modules[vmlinux_module_name]
+        vmlinux_layer = self.context.layers[vmlinux.layer_name]
+        tar_buffer = BytesIO()
+        tar = tarfile.open(
+            fileobj=tar_buffer,
+            mode=f"w:{self.config['compression_format']}",
+        )
+        # Set a unique timestamp for all extracted files
+        mtime = time.time()
+
+        inodes_iter = Files.get_inodes(
+            context=self.context,
+            vmlinux_module_name=vmlinux_module_name,
+            follow_symlinks=False,
+        )
+
+        # Prefix paths with the superblock UUID's to prevent overlaps.
+        # Switch to device major and device minor for older kernels (< 2.6.39-rc1).
+        uuid_as_prefix = vmlinux.get_type("super_block").has_member("s_uuid")
+        if not uuid_as_prefix:
+            vollog.warning(
+                "super_block struct does not support s_uuid attribute. Consequently, level 0 directories won't refer to the superblock uuid's, but to its device_major:device_minor numbers."
+            )
+
+        visited_paths = seen_prefixes = set()
+        for inode_in in inodes_iter:
+
+            # Code is slightly duplicated here with the if-block below.
+            # However this prevents unneeded tar manipulation if fifo
+            # or sock inodes come through for example.
+            if not (
+                inode_in.inode.is_reg or inode_in.inode.is_dir or inode_in.inode.is_link
+            ):
+                continue
+
+            if not inode_in.path.startswith("/"):
+                vollog.debug(
+                    f'Skipping processing of potentially smeared "{inode_in.path}" inode name as it does not starts with a "/".'
+                )
+                continue
+
+            # Construct the output path
+            if uuid_as_prefix:
+                prefix = f"/{inode_in.superblock.uuid}"
+            else:
+                prefix = f"/{inode_in.superblock.major}:{inode_in.superblock.minor}"
+            prefixed_path = prefix + inode_in.path
+
+            # Sanity check for already processed paths
+            if prefixed_path in visited_paths:
+                vollog.log(
+                    constants.LOGLEVEL_VV,
+                    f'Already processed prefixed inode path: "{prefixed_path}".',
+                )
+                continue
+            elif prefix not in seen_prefixes:
+                self._tar_add_dir(tar, prefix, mtime)
+                seen_prefixes.add(prefix)
+
+            visited_paths.add(prefixed_path)
+            extracted_file_size = renderers.NotApplicableValue()
+
+            # Inodes parent directory is yielded first, which
+            # ensures that a file parent path will exist beforehand.
+            # tarfile will take care of creating it anyway.
+            if inode_in.inode.is_reg:
+                extracted_file_size = self._tar_add_reg_inode(
+                    self.context,
+                    vmlinux_layer.name,
+                    tar,
+                    inode_in,
+                    prefix,
+                    mtime,
+                )
+            elif inode_in.inode.is_dir:
+                self._tar_add_dir(tar, prefixed_path, mtime)
+            elif (
+                inode_in.inode.is_link
+                and inode_in.inode.has_member("i_link")
+                and inode_in.inode.i_link
+                and inode_in.inode.i_link.is_readable()
+            ):
+                symlink_dest = inode_in.inode.i_link.dereference().cast(
+                    "string", max_length=255, encoding="utf-8", errors="replace"
+                )
+                self._tar_add_lnk(tar, inode_in.path, symlink_dest, prefix, mtime)
+                # Set path to a user friendly representation before yielding
+                inode_in.path = InodeUser.format_symlink(inode_in.path, symlink_dest)
+            else:
+                continue
+
+            inode_out = inode_in.to_user(vmlinux_layer)
+            yield (0, astuple(inode_out) + (extracted_file_size,))
+
+        tar.close()
+        tar_buffer.seek(0)
+        output_filename = f"recovered_fs.tar.{self.config['compression_format']}"
+        with self.open(output_filename) as f:
+            f.write(tar_buffer.getvalue())
+
+    def run(self):
+        headers = [
+            ("SuperblockAddr", format_hints.Hex),
+            ("MountPoint", str),
+            ("Device", str),
+            ("InodeNum", int),
+            ("InodeAddr", format_hints.Hex),
+            ("FileType", str),
+            ("InodePages", int),
+            ("CachedPages", int),
+            ("FileMode", str),
+            ("AccessTime", datetime.datetime),
+            ("ModificationTime", datetime.datetime),
+            ("ChangeTime", datetime.datetime),
+            ("FilePath", str),
+            ("InodeSize", int),
+            ("Recovered FileSize", int),
+        ]
+
+        return renderers.TreeGrid(
+            headers, Files.format_fields_with_headers(headers, self._generator())
+        )

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

@@ -10,6 +10,7 @@
 import stat
 import datetime
 import socket as socket_module
+import uuid
 from typing import (
     Generator,
     Iterable,
@@ -1057,14 +1058,28 @@ class super_block(objects.StructType):
         SB_LAZYTIME: "lazytime",
     }
 
-    @property
+    @functools.cached_property
     def major(self) -> int:
         return self.s_dev >> self.MINORBITS
 
-    @property
+    @functools.cached_property
     def minor(self) -> int:
         return self.s_dev & ((1 << self.MINORBITS) - 1)
 
+    @functools.cached_property
+    def uuid(self) -> str:
+        if not self.has_member("s_uuid"):
+            raise AttributeError(
+                "super_block struct does not support s_uuid direct attribute access, probably indicating a kernel version < 2.6.39-rc1."
+            )
+
+        if self.s_uuid.has_member("b"):
+            uuid_as_ints = self.s_uuid.b
+        else:
+            uuid_as_ints = self.s_uuid
+
+        return str(uuid.UUID(bytes=bytes(uuid_as_ints)))
+
     def get_flags_access(self) -> str:
         return "ro" if self.s_flags & self.SB_RDONLY else "rw"