volatility3/volatility/framework/plugins/mac/kevents.py

# This file is opyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#

from typing import Iterable, Callable, Tuple

from volatility.framework import renderers, interfaces, exceptions, contexts
from volatility.framework.configuration import requirements
from volatility.framework.objects import utility
from volatility.framework.symbols import mac
from volatility.plugins.mac import pslist


class Kevents(interfaces.plugins.PluginInterface):
    """ Lists event handlers registered by processes """

    _required_framework_version = (2, 0, 0)

    event_types = {
        1: "EVFILT_READ",
        2: "EVFILT_WRITE",
        3: "EVFILT_AIO",
        4: "EVFILT_VNODE",
        5: "EVFILT_PROC",
        6: "EVFILT_SIGNAL",
        7: "EVFILT_TIMER",
        8: "EVFILT_MACHPORT",
        9: "EVFILT_FS",
        10: "EVFILT_USER",
        12: "EVFILT_VM"
    }

    vnode_filters = [("NOTE_DELETE", 1), ("NOTE_WRITE", 2), ("NOTE_EXTEND", 4), ("NOTE_ATTRIB", 8), ("NOTE_LINK", 0x10),
                     ("NOTE_RENAME", 0x20), ("NOTE_REVOKE", 0x40)]

    proc_filters = [("NOTE_EXIT", 0x80000000), ("NOTE_EXITSTATUS", 0x04000000), ("NOTE_FORK", 0x40000000),
                    ("NOTE_EXEC", 0x20000000), ("NOTE_SIGNAL", 0x08000000), ("NOTE_REAP", 0x10000000)]

    timer_filters = [("NOTE_SECONDS", 1), ("NOTE_USECONDS", 2), ("NOTE_NSECONDS", 4), ("NOTE_ABSOLUTE", 8)]

    all_filters = {
        4: vnode_filters,  # EVFILT_VNODE
        5: proc_filters,  # EVFILT_PROC
        7: timer_filters  # EVFILT_TIMER
    }

    @classmethod
    def get_requirements(cls):
        return [
            requirements.TranslationLayerRequirement(name = 'primary',
                                                     description = 'Memory layer for the kernel',
                                                     architectures = ["Intel32", "Intel64"]),
            requirements.SymbolTableRequirement(name = "darwin", description = "Mac kernel"),
            requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
            requirements.VersionRequirement(name = 'macutils', component = mac.MacUtilities, version = (1, 2, 0)),
            requirements.ListRequirement(name = 'pid',
                                         description = 'Filter on specific process IDs',
                                         element_type = int,
                                         optional = True)
        ]

    def _parse_flags(self, filter_index, filter_flags):
        if filter_flags == 0 or filter_index not in self.all_filters:
            return ""

        context = []

        filters = self.all_filters[filter_index]
        for flag, index in filters:
            if filter_flags & index == index:
                context.append(flag)

        return ",".join(context)

    @classmethod
    def _walk_klist_array(cls, kernel, fdp, array_pointer_member, array_size_member):
        """
        Convience wrapper for walking an array of lists of kernel events
        Handles invalid address references
        """
        try:
            klist_array_pointer = getattr(fdp, array_pointer_member)
            array_size = getattr(fdp, array_size_member)

            klist_array = kernel.object(object_type = "array",
                                        offset = klist_array_pointer,
                                        count = array_size + 1,
                                        subtype = kernel.get_type("klist"))

        except exceptions.InvalidAddressException:
            return

        for klist in klist_array:
            for kn in mac.MacUtilities.walk_slist(klist, "kn_link"):
                yield kn

    @classmethod
    def _get_task_kevents(cls, kernel, task):
        """
        Enumerates event filters per task.
        Uses smear-safe APIs throughout as these data structures
        see a signifcant amount of smear
        """
        fdp = task.p_fd

        for kn in cls._walk_klist_array(kernel, fdp, "fd_knlist", "fd_knlistsize"):
            yield kn

        for kn in cls._walk_klist_array(kernel, fdp, "fd_knhash", "fd_knhashmask"):
            yield kn

        try:
            p_klist = task.p_klist
        except exceptions.InvalidAddressException:
            return

        for kn in mac.MacUtilities.walk_slist(p_klist, "kn_link"):
            yield kn

    @classmethod
    def list_kernel_events(cls,
                           context: interfaces.context.ContextInterface,
                           layer_name: str,
                           darwin_symbols: str,
                           filter_func: Callable[[int], bool] = lambda _: False) -> \
            Iterable[Tuple[interfaces.objects.ObjectInterface,
                           interfaces.objects.ObjectInterface,
                           interfaces.objects.ObjectInterface]]:
        """
        Returns the kernel event filters registered

        Return values:
            A tuple of 3 elements:
                1) The name of the process that registered the filter
                2) The process ID of the process that registered the filter
                3) The object of the associated kernel event filter
        """
        kernel = contexts.Module(context, darwin_symbols, layer_name, 0)

        list_tasks = pslist.PsList.get_list_tasks(pslist.PsList.pslist_methods[0])

        for task in list_tasks(context, layer_name, darwin_symbols, filter_func):
            task_name = utility.array_to_string(task.p_comm)
            pid = task.p_pid

            for kn in cls._get_task_kevents(kernel, task):
                yield task_name, pid, kn

    def _generator(self):
        filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))

        for task_name, pid, kn in self.list_kernel_events(self.context,
                                                          self.config['primary'],
                                                          self.config['darwin'],
                                                          filter_func = filter_func):

            filter_index = kn.kn_kevent.filter * -1
            if filter_index in self.event_types:
                filter_name = self.event_types[filter_index]
            else:
                continue

            try:
                ident = kn.kn_kevent.ident
            except exceptions.InvalidAddressException:
                continue

            context = self._parse_flags(filter_index, kn.kn_sfflags)

            yield (0, (pid, task_name, ident, filter_name, context))

    def run(self):
        return renderers.TreeGrid([("PID", int), ("Process", str), ("Ident", int), ("Filter", str), ("Context", str)],
                                  self._generator())