Merge pull request #1317 from gcmoreira/linux_boottime_support

Linux - Boottime support

volatility3/framework/constants/linux/__init__.py

@@ -304,6 +304,7 @@ class ELF_CLASS(IntEnum):
     ELFCLASS64 = 2
 
 
+# PTrace
 PT_OPT_FLAG_SHIFT = 3
 
 PTRACE_EVENT_FORK = 1
@@ -341,6 +342,10 @@ class PT_FLAGS(Flag):
         return str(self).replace(self.__class__.__name__ + ".", "")
 
 
+# Boot time
+NSEC_PER_SEC = 1e9
+
+
 # Valid sizes for modules. Note that the Linux kernel does not define these values; they
 # are based on empirical observations of typical memory allocations for kernel modules.
 # We use this to verify that the found module falls within reasonable limits.

volatility3/framework/plugins/linux/boottime.py

@@ -0,0 +1,98 @@
+# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+import datetime
+from typing import List, Tuple, Iterable
+
+
+from volatility3.framework import interfaces, renderers
+from volatility3.framework.configuration import requirements
+from volatility3.plugins import timeliner
+from volatility3.plugins.linux import pslist
+
+
+class Boottime(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
+    """Shows the time the system was started"""
+
+    _required_framework_version = (2, 11, 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, 3, 0)
+            ),
+        ]
+
+    @classmethod
+    def get_time_namespaces_bootime(
+        cls,
+        context: interfaces.context.ContextInterface,
+        vmlinux_module_name: str,
+    ) -> Iterable[Tuple[int, int, int, str, datetime.datetime]]:
+        """Enumerates tasks' boot times based on their time namespaces.
+
+        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
+            pids: Pid list
+            unique: Filter unique time namespaces
+
+        Yields:
+            A tuple with the fields to show in the plugin output.
+        """
+        time_namespace_ids = set()
+        for task in pslist.PsList.list_tasks(context, vmlinux_module_name):
+            time_namespace_id = task.get_time_namespace_id()
+            # If it cannot get the time namespace i.e. kernels < 5.6, this still works
+            # using None to just get the first tasks
+            if time_namespace_id in time_namespace_ids:
+                continue
+            time_namespace_ids.add(time_namespace_id)
+            boottime = task.get_boottime(root_time_namespace=False)
+
+            fields = (
+                time_namespace_id,
+                boottime,
+            )
+            yield fields
+
+    def _generator(self):
+        for (
+            time_namespace_id,
+            boottime,
+        ) in self.get_time_namespaces_bootime(
+            self.context,
+            self.config["kernel"],
+        ):
+            fields = [
+                time_namespace_id or renderers.NotAvailableValue(),
+                boottime,
+            ]
+            yield 0, fields
+
+    def generate_timeline(self):
+        for (
+            time_namespace_id,
+            boottime,
+        ) in self.get_time_namespaces_bootime(
+            self.context,
+            self.config["kernel"],
+        ):
+            description = f"System boot time for time namespace {time_namespace_id}"
+
+            yield description, timeliner.TimeLinerType.CREATED, boottime
+
+    def run(self):
+        columns = [
+            ("TIME NS", int),
+            ("Boot Time", datetime.datetime),
+        ]
+        return renderers.TreeGrid(columns, self._generator())

volatility3/framework/plugins/linux/pslist.py

@@ -1,6 +1,7 @@
 # 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
 #
+import datetime
 from typing import Any, Callable, Iterable, List, Tuple
 
 from volatility3.framework import interfaces, renderers
@@ -9,15 +10,16 @@ from volatility3.framework.objects import utility
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.linux.extensions import elf
+from volatility3.plugins import timeliner
 from volatility3.plugins.linux import elfs
 
 
-class PsList(interfaces.plugins.PluginInterface):
+class PsList(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
     """Lists the processes present in a particular linux memory image."""
 
     _required_framework_version = (2, 0, 0)
 
-    _version = (2, 2, 1)
+    _version = (2, 3, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -81,7 +83,7 @@ class PsList(interfaces.plugins.PluginInterface):
     @classmethod
     def get_task_fields(
         cls, task: interfaces.objects.ObjectInterface, decorate_comm: bool = False
-    ) -> Tuple[int, int, int, str]:
+    ) -> Tuple[int, int, int, int, str, datetime.datetime]:
         """Extract the fields needed for the final output
 
         Args:
@@ -96,13 +98,14 @@ class PsList(interfaces.plugins.PluginInterface):
         tid = task.pid
         ppid = task.parent.tgid if task.parent else 0
         name = utility.array_to_string(task.comm)
+        start_time = task.get_create_time()
         if decorate_comm:
             if task.is_kernel_thread:
                 name = f"[{name}]"
             elif task.is_user_thread:
                 name = f"{{{name}}}"
 
-        task_fields = (task.vol.offset, pid, tid, ppid, name)
+        task_fields = (task.vol.offset, pid, tid, ppid, name, start_time)
         return task_fields
 
     def _get_file_output(self, task: interfaces.objects.ObjectInterface) -> str:
@@ -177,7 +180,9 @@ class PsList(interfaces.plugins.PluginInterface):
             else:
                 file_output = "Disabled"
 
-            offset, pid, tid, ppid, name = self.get_task_fields(task, decorate_comm)
+            offset, pid, tid, ppid, name, creation_time = self.get_task_fields(
+                task, decorate_comm
+            )
 
             yield 0, (
                 format_hints.Hex(offset),
@@ -185,6 +190,7 @@ class PsList(interfaces.plugins.PluginInterface):
                 tid,
                 ppid,
                 name,
+                creation_time or renderers.NotAvailableValue(),
                 file_output,
             )
 
@@ -233,8 +239,23 @@ class PsList(interfaces.plugins.PluginInterface):
             ("TID", int),
             ("PPID", int),
             ("COMM", str),
+            ("CREATION TIME", datetime.datetime),
             ("File output", str),
         ]
         return renderers.TreeGrid(
             columns, self._generator(filter_func, include_threads, decorate_comm, dump)
         )
+
+    def generate_timeline(self):
+        pids = self.config.get("pid")
+        filter_func = self.create_pid_filter(pids)
+        for task in self.list_tasks(
+            self.context, self.config["kernel"], filter_func, include_threads=True
+        ):
+            offset, user_pid, user_tid, _user_ppid, name, creation_time = (
+                self.get_task_fields(task)
+            )
+
+            description = f"Process {user_pid}/{user_tid} {name} ({offset})"
+
+            yield (description, timeliner.TimeLinerType.CREATED, creation_time)

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

@@ -2,12 +2,13 @@
 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
 #
 
+import abc
 import collections.abc
 import logging
 import functools
 import binascii
 import stat
-from datetime import datetime
+import datetime
 import socket as socket_module
 from typing import Generator, Iterable, Iterator, Optional, Tuple, List, Union, Dict
 
@@ -19,6 +20,7 @@ from volatility3.framework.objects import utility
 from volatility3.framework.symbols import generic, linux, intermed
 from volatility3.framework.symbols.linux.extensions import elf
 
+
 vollog = logging.getLogger(__name__)
 
 # Keep these in a basic module, to prevent import cycles when symbol providers require them
@@ -444,6 +446,197 @@ class task_struct(generic.GenericIntelProcess):
             else None
         )
 
+    def _get_task_start_time(self) -> datetime.timedelta:
+        """Returns the task's monotonic start_time as a timedelta.
+
+        Returns:
+            The task's start time as a timedelta object.
+        """
+        for member_name in ("start_boottime", "real_start_time", "start_time"):
+            if self.has_member(member_name):
+                start_time_obj = self.member(member_name)
+                start_time_obj_type = start_time_obj.vol.type_name
+                start_time_obj_type_name = start_time_obj_type.split(constants.BANG)[1]
+                if start_time_obj_type_name != "timespec":
+                    # kernels >= 3.17 real_start_time and start_time are u64
+                    # kernels >= 5.5 uses start_boottime which is also a u64
+                    start_time = Timespec64Concrete.new_from_nsec(start_time_obj)
+                else:
+                    # kernels < 3.17 real_start_time and start_time are timespec
+                    start_time = Timespec64Concrete.new_from_timespec(start_time_obj)
+
+                # This is relative to the boot time so it makes sense to be a timedelta.
+                return start_time.to_timedelta()
+
+        raise AttributeError("Unsupported task_struct start_time member")
+
+    def get_time_namespace(self) -> Optional[interfaces.objects.ObjectInterface]:
+        """Returns the task's time namespace"""
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        if not self.has_member("nsproxy"):
+            # kernels < 2.6.19: ab516013ad9ca47f1d3a936fa81303bfbf734d52
+            return None
+
+        if not vmlinux.get_type("nsproxy").has_member("time_ns"):
+            # kernels < 5.6 769071ac9f20b6a447410c7eaa55d1a5233ef40c
+            return None
+
+        return self.nsproxy.time_ns
+
+    def get_time_namespace_id(self) -> int:
+        """Returns the task's time namespace ID."""
+        time_ns = self.get_time_namespace()
+        if not time_ns:
+            # kernels < 5.6
+            return None
+
+        # We are good. ns_common (ns) was introduced in kernels 3.19. So by the time the
+        # time namespace was added in kernels 5.6, it already included the ns member.
+        return time_ns.ns.inum
+
+    def _get_time_namespace_offsets(
+        self,
+    ) -> Optional[interfaces.objects.ObjectInterface]:
+        """Returns the time offsets from the task's time namespace."""
+        time_ns = self.get_time_namespace()
+        if not time_ns:
+            # kernels < 5.6
+            return None
+
+        if not time_ns.has_member("offsets"):
+            # kernels < 5.6 af993f58d69ee9c1f421dfc87c3ed231c113989c
+            return None
+
+        return time_ns.offsets
+
+    def get_time_namespace_monotonic_offset(
+        self,
+    ) -> Optional[interfaces.objects.ObjectInterface]:
+        """Gets task's time namespace monotonic offset
+
+        Returns:
+            a kernel's timespec64 object with the monotonic offset
+        """
+        time_namespace_offsets = self._get_time_namespace_offsets()
+        if not time_namespace_offsets:
+            return None
+
+        return time_namespace_offsets.monotonic
+
+    def _get_time_namespace_boottime_offset(
+        self,
+    ) -> Optional[interfaces.objects.ObjectInterface]:
+        """Gets task's time namespace boottime offset
+
+        Returns:
+            a kernel's timespec64 object with the boottime offset
+        """
+        time_namespace_offsets = self._get_time_namespace_offsets()
+        if not time_namespace_offsets:
+            return None
+
+        return time_namespace_offsets.boottime
+
+    def _get_boottime_raw(self) -> "Timespec64Concrete":
+        """Returns the boot time in a Timespec64Concrete object."""
+
+        vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
+        if vmlinux.has_symbol("tk_core"):
+            # kernels >= 3.17 | tk_core | 3fdb14fd1df70325e1e91e1203a699a4803ed741
+            tk_core = vmlinux.object_from_symbol("tk_core")
+            timekeeper = tk_core.timekeeper
+            if not timekeeper.offs_real.has_member("tv64"):
+                # kernels >= 4.10 - Tested on Ubuntu 6.8.0-41
+                boottime_nsec = timekeeper.offs_real - timekeeper.offs_boot
+            else:
+                # 3.17 <= kernels < 4.10 - Tested on Ubuntu 4.4.0-142
+                boottime_nsec = timekeeper.offs_real.tv64 - timekeeper.offs_boot.tv64
+            return Timespec64Concrete.new_from_nsec(boottime_nsec)
+
+        elif vmlinux.has_symbol("timekeeper") and vmlinux.get_type(
+            "timekeeper"
+        ).has_member("wall_to_monotonic"):
+            # 3.4 <= kernels < 3.17 - Tested on Ubuntu 3.13.0-185
+            timekeeper = vmlinux.object_from_symbol("timekeeper")
+
+            # timekeeper.wall_to_monotonic is timespec
+            boottime = Timespec64Concrete.new_from_timespec(
+                timekeeper.wall_to_monotonic
+            )
+
+            boottime += timekeeper.total_sleep_time
+
+            return boottime.negate()
+
+        elif vmlinux.has_symbol("wall_to_monotonic"):
+            # kernels < 3.4 - Tested on Debian7 3.2.0-4 (3.2.57-3+deb7u2)
+            wall_to_monotonic = vmlinux.object_from_symbol("wall_to_monotonic")
+            boottime = Timespec64Concrete.new_from_timespec(wall_to_monotonic)
+            if vmlinux.has_symbol("total_sleep_time"):
+                # 2.6.23 <= kernels < 3.4 7c3f1a573237b90ef331267260358a0ec4ac9079
+                total_sleep_time = vmlinux.object_from_symbol("total_sleep_time")
+                full_type_name = total_sleep_time.vol.type_name
+                type_name = full_type_name.split(constants.BANG)[1]
+                if type_name == "timespec":
+                    # kernels >= 2.6.32 total_sleep_time is a timespec
+                    boottime += total_sleep_time
+                else:
+                    # kernels < 2.6.32 total_sleep_time is an unsigned long as seconds
+                    boottime.tv_sec += total_sleep_time
+
+            return boottime.negate()
+
+        raise exceptions.VolatilityException("Unsupported")
+
+    def get_boottime(self, root_time_namespace: bool = True) -> datetime.datetime:
+        """Returns the boot time in UTC as a datetime.
+
+        Args:
+            root_time_namespace: If True, it returns the boot time as seen from the root
+                time namespace. Otherwise, it returns the boot time relative to the
+                task's time namespace.
+
+        Returns:
+            A datetime with the UTC boot time.
+        """
+        boottime = self._get_boottime_raw()
+        if not boottime:
+            return None
+
+        if not root_time_namespace:
+            # Shift boot timestamp according to the task's time namespace offset
+            boottime_offset_timespec = self._get_time_namespace_boottime_offset()
+            if boottime_offset_timespec:
+                # Time namespace support is from kernels 5.6
+                boottime -= boottime_offset_timespec
+
+        return boottime.to_datetime()
+
+    def get_create_time(self) -> datetime.datetime:
+        """Retrieves the task's start time from its time namespace.
+        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
+            task: A reference task
+
+        Returns:
+            A datetime with task's start time
+        """
+        # Typically, we want to see the creation time seen from the root time namespace
+        boottime = self.get_boottime(root_time_namespace=True)
+
+        # The kernel exports only tv_sec to procfs, see kernel's show_stat().
+        # This means user-space tools, like those in the procps package (e.g., ps, top, etc.),
+        # only use the boot time seconds to compute dates relatives to this.
+        boottime = boottime.replace(microsecond=0)
+
+        task_start_time_timedelta = self._get_task_start_time()
+
+        # NOTE: Do NOT apply the task's time namespace offsets here. While the kernel uses
+        # timens_add_boottime_ns(), it's not needed here since we're seeing it from the
+        # root time namespace, not within the task's own time namespace
+        return boottime + task_start_time_timedelta
+
 
 class fs_struct(objects.StructType):
     def get_root_dentry(self):
@@ -2004,12 +2197,118 @@ class kernel_cap_t(kernel_cap_struct):
         return cap_value & self.get_kernel_cap_full()
 
 
-class timespec64(objects.StructType):
-    def to_datetime(self) -> datetime:
-        """Returns the respective aware datetime"""
+class Timespec64Abstract(abc.ABC):
+    """Abstract class to handle all required timespec64 operations, convertions and
+    adjustments."""
 
-        dt = conversion.unixtime_to_datetime(self.tv_sec + self.tv_nsec / 1e9)
-        return dt
+    @classmethod
+    def new_from_timespec(cls, other) -> "Timespec64Concrete":
+        """Creates a new instance from an Timespec64Abstract subclass object"""
+        if not isinstance(other, Timespec64Abstract):
+            raise TypeError("Requires an object subclass of Timespec64Abstract")
+
+        tv_sec = int(other.tv_sec)
+        tv_nsec = int(other.tv_nsec)
+        return Timespec64Concrete(tv_sec=tv_sec, tv_nsec=tv_nsec)
+
+    @classmethod
+    def new_from_nsec(cls, nsec) -> "Timespec64Concrete":
+        """Creates a new instance from an integer in nanoseconds"""
+
+        # Based on ns_to_timespec64()
+        if nsec > 0:
+            tv_sec = nsec // linux_constants.NSEC_PER_SEC
+            tv_nsec = nsec % linux_constants.NSEC_PER_SEC
+        elif nsec < 0:
+            tv_sec = -((-nsec - 1) // linux_constants.NSEC_PER_SEC) - 1
+            rem = (-nsec - 1) % linux_constants.NSEC_PER_SEC
+            tv_nsec = linux_constants.NSEC_PER_SEC - rem - 1
+        else:
+            tv_sec = tv_nsec = 0
+
+        return Timespec64Concrete(tv_sec=tv_sec, tv_nsec=tv_nsec)
+
+    def to_datetime(self) -> datetime.datetime:
+        """Converts this Timespec64Abstract subclass object to a UTC aware datetime"""
+
+        # pylint: disable=E1101
+        return conversion.unixtime_to_datetime(
+            self.tv_sec + self.tv_nsec / linux_constants.NSEC_PER_SEC
+        )
+
+    def to_timedelta(self) -> datetime.timedelta:
+        """Converts this Timespec64Abstract subclass object to timedelta"""
+        # pylint: disable=E1101
+        return datetime.timedelta(
+            seconds=self.tv_sec + self.tv_nsec / linux_constants.NSEC_PER_SEC
+        )
+
+    def __add__(self, other) -> "Timespec64Concrete":
+        """Returns a new Timespec64Concrete object that sums the current values with those
+        in the timespec argument"""
+        if not isinstance(other, Timespec64Abstract):
+            raise TypeError("Requires an object subclass of Timespec64Abstract")
+
+        # pylint: disable=E1101
+        result = Timespec64Concrete(
+            tv_sec=self.tv_sec + other.tv_sec,
+            tv_nsec=self.tv_nsec + other.tv_nsec,
+        )
+
+        result.normalize()
+
+        return result
+
+    def __sub__(self, other) -> "Timespec64Concrete":
+        """Returns a new Timespec64Abstract object that subtracts the values in the timespec
+        argument from the current object's values"""
+        if not isinstance(other, Timespec64Abstract):
+            raise TypeError("Requires an object subclass of Timespec64Abstract")
+
+        return self + other.negate()
+
+    def negate(self) -> "Timespec64Concrete":
+        """Returns a new Timespec64Concrete object with the values of the current object negated"""
+        # pylint: disable=E1101
+        result = Timespec64Concrete(
+            tv_sec=-self.tv_sec,
+            tv_nsec=-self.tv_nsec,
+        )
+
+        result.normalize()
+
+        return result
+
+    def normalize(self):
+        """Normalize any overflow in tv_sec and tv_nsec."""
+        # Based on kernel's set_normalized_timespec64()
+
+        # pylint: disable=E1101
+        while self.tv_nsec >= linux_constants.NSEC_PER_SEC:
+            self.tv_nsec -= linux_constants.NSEC_PER_SEC
+            self.tv_sec += 1
+
+        while self.tv_nsec < 0:
+            self.tv_nsec += linux_constants.NSEC_PER_SEC
+            self.tv_sec -= 1
+
+
+class Timespec64Concrete(Timespec64Abstract):
+    """Handle all required timespec64 operations, convertions and adjustments.
+    This is used to dynamically create timespec64-like objects, each with its own variables
+    and the same methods as a timespec64 object extension.
+    """
+
+    def __init__(self, tv_sec=0, tv_nsec=0):
+        self.tv_sec = tv_sec
+        self.tv_nsec = tv_nsec
+
+
+class timespec64(Timespec64Abstract, objects.StructType):
+    """Handle all required timespec64 operations, convertions and adjustments.
+    This works as an extension of the timespec64 object while maintaining the same methods
+    as a Timespec64Concrete object.
+    """
 
 
 class inode(objects.StructType):
@@ -2081,7 +2380,7 @@ class inode(objects.StructType):
         else:
             return None
 
-    def _time_member_to_datetime(self, member) -> datetime:
+    def _time_member_to_datetime(self, member) -> datetime.datetime:
         if self.has_member(f"{member}_sec") and self.has_member(f"{member}_nsec"):
             # kernels >= 6.11 it's i_*_sec -> time64_t and i_*_nsec -> u32
             # Ref Linux commit 3aa63a569c64e708df547a8913c84e64a06e7853
@@ -2100,7 +2399,7 @@ class inode(objects.StructType):
                 "Unsupported kernel inode type implementation"
             )
 
-    def get_access_time(self) -> datetime:
+    def get_access_time(self) -> datetime.datetime:
         """Returns the inode's last access time
         This is updated when inode contents are read
 
@@ -2109,7 +2408,7 @@ class inode(objects.StructType):
         """
         return self._time_member_to_datetime("i_atime")
 
-    def get_modification_time(self) -> datetime:
+    def get_modification_time(self) -> datetime.datetime:
         """Returns the inode's last modification time
         This is updated when the inode contents change
 
@@ -2119,7 +2418,7 @@ class inode(objects.StructType):
 
         return self._time_member_to_datetime("i_mtime")
 
-    def get_change_time(self) -> datetime:
+    def get_change_time(self) -> datetime.datetime:
         """Returns the inode's last change time
         This is updated when the inode metadata changes