Add svclist and svcdiff plugins. Make svcscan more modular to support inheritance and cleaner code

volatility3/framework/plugins/windows/svcdiff.py

@@ -0,0 +1,74 @@
+# 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
+#
+
+# This module attempts to locate skeleton-key like function hooks.
+# It does this by locating the CSystems array through a variety of methods,
+# and then validating the entry for RC4 HMAC (0x17 / 23)
+#
+# For a thorough walkthrough on how the R&D was performed to develop this plugin,
+# please see our blogpost here:
+#
+# https://volatility-labs.blogspot.com/2021/10/memory-forensics-r-illustrated.html
+
+import logging
+
+from volatility3.framework import symbols
+from volatility3.framework.configuration import requirements
+from volatility3.plugins.windows import svclist, svcscan
+from volatility3.framework.symbols.windows import versions
+
+vollog = logging.getLogger(__name__)
+
+class SvcDiff(svclist.SvcList, svcscan.SvcScan):
+    """Compares services found through list walking versus scanning to find rootkits"""
+
+    _required_framework_version = (2, 4, 0)
+
+    @classmethod
+    def get_requirements(cls):
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.ModuleRequirement(
+                name="kernel",
+                description="Windows kernel",
+                architectures=["Intel32", "Intel64"],
+            ),
+            requirements.VersionRequirement(
+                name="svclist", component=svclist.SvcList, version=(1, 0, 0)
+            ),
+            requirements.VersionRequirement(
+                name="svcscan", component=svcscan.SvcScan, version=(2, 0, 0)
+            ),
+        ]
+
+    def _generator(self):
+        """
+        Finds services by walking the services.exe list on supported Windows 10 versions
+        """
+        kernel = self.context.modules[self.config["kernel"]]
+
+        if not symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name) or \
+           not versions.is_win10_15063_or_later(context=self.context, symbol_table=kernel.symbol_table_name):
+            vollog.info("This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples")
+            return
+
+        from_scan = set()
+        from_list = set()
+        records = {}
+        
+        service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info()
+
+        # collect unique service names from scanning
+        for service in self.service_scan(service_table_name, service_binary_dll_map, filter_func):
+            from_scan.add(service[6])
+            records[service[6]] = service
+
+        # collect services from listing walking
+        for service in self.service_list(service_table_name, service_binary_dll_map, filter_func):
+            from_list.add(service[6])
+
+        # report services found from scanning but not list walking
+        for hidden_service in from_scan-from_list:
+            yield (0, records[hidden_service])
+

volatility3/framework/plugins/windows/svclist.py

@@ -0,0 +1,86 @@
+# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
+# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
+#
+
+import logging
+
+from typing import List
+
+from volatility3.framework import interfaces, exceptions,  symbols
+from volatility3.framework.configuration import requirements
+from volatility3.framework.symbols.windows import versions
+from volatility3.plugins.windows import svcscan, pslist
+from volatility3.framework.layers import scanners
+
+vollog = logging.getLogger(__name__)
+
+
+class SvcList(svcscan.SvcScan):
+    """Lists services contained with the services.exe doubly linked list of services"""
+
+    _version = (1, 0, 0)
+
+    @classmethod
+    def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+        # Since we're calling the plugin, make sure we have the plugin's requirements
+        return [
+            requirements.PluginRequirement(
+                name="svcscan", plugin=svcscan.SvcScan, version=(2, 0, 0)
+            ),
+        ]
+
+    def _get_exe_range(self, proc):
+        """
+        Returns a tuple of starting,ending address for
+        the VAD containing services.exe
+        """
+
+        vad_root = proc.get_vad_root()
+        for vad in vad_root.traverse():
+            filename = vad.get_file_name()
+            if isinstance(filename, str) and filename.lower().endswith("\\services.exe"):
+                return [(vad.get_start(), vad.get_size())]
+
+        return None
+
+    def service_list(self, service_table_name, service_binary_dll_map, filter_func):
+        kernel = self.context.modules[self.config["kernel"]]
+
+        if not symbols.symbol_table_is_64bit(self.context, kernel.symbol_table_name) or \
+           not versions.is_win10_15063_or_later(context=self.context, symbol_table=kernel.symbol_table_name):
+            vollog.info("This plugin only supports Windows 10 version 15063+ 64bit Windows memory samples")
+            return
+
+        for proc in pslist.PsList.list_processes(
+            context=self.context,
+            layer_name=kernel.layer_name,
+            symbol_table=kernel.symbol_table_name,
+            filter_func=filter_func,
+        ):
+            try:
+                layer_name = proc.add_process_layer()
+            except exceptions.InvalidAddressException:
+                vollog.warning("Unable to access memory of services.exe running with PID: {}".format(proc.UniqueProcessId))
+                continue
+
+            layer = self.context.layers[layer_name]
+
+            exe_range = self._get_exe_range(proc)
+            if not exe_range:
+                vollog.warning("Could not find the application executable VAD for services.exe. Unable to proceed.")
+                continue
+
+            for offset in layer.scan(
+                context=self.context,
+                scanner=scanners.BytesScanner(needle = b"Sc27"),
+                sections=exe_range,
+                ):
+                for record in self.enumerate_vista_or_later_header(service_table_name, service_binary_dll_map, layer_name, offset):
+                    yield record
+
+    def _generator(self):
+        service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info()
+        
+        for record in self.service_list(service_table_name, service_binary_dll_map, filter_func):
+            yield (0, record)
+

volatility3/framework/plugins/windows/svcscan.py

@@ -19,7 +19,7 @@ from volatility3.framework.layers import scanners
 from volatility3.framework.renderers import format_hints
 from volatility3.framework.symbols import intermed
 from volatility3.framework.symbols.windows import versions
-from volatility3.framework.symbols.windows.extensions import services
+from volatility3.framework.symbols.windows.extensions import services as services_types
 from volatility3.plugins.windows import poolscanner, pslist, vadyarascan
 from volatility3.plugins.windows.registry import hivelist
 
@@ -140,7 +140,7 @@ class SvcScan(interfaces.plugins.PluginInterface):
             config_path,
             os.path.join("windows", "services"),
             symbol_filename,
-            class_types=services.class_types,
+            class_types=services_types.class_types,
             native_types=native_types,
         )
 
@@ -232,28 +232,44 @@ class SvcScan(interfaces.plugins.PluginInterface):
             for service_key in services
         }
 
-    def _generator(self):
+    def enumerate_vista_or_later_header(
+        self,
+        service_table_name,
+        service_binary_dll_map,
+        proc_layer_name,
+        offset
+    ):
+        if offset % 8:
+            return
+
+        service_header = self.context.object(
+            service_table_name + constants.BANG + "_SERVICE_HEADER",
+            offset=offset,
+            layer_name=proc_layer_name,
+        )
+
+        if not service_header.is_valid():
+            return
+
+        # since we walk the s-list backwards, if we've seen
+        # an object, then we've also seen all objects that
+        # exist before it, thus we can break at that time.
+        for service_record in service_header.ServiceRecord.traverse():
+            service_info = service_binary_dll_map.get(
+                service_record.get_name(),
+                ServiceBinaryInfo(
+                    renderers.UnreadableValue(), renderers.UnreadableValue()
+                ),
+            )
+            yield self.get_record_tuple(service_record, service_info)
+
+    def service_scan(self, service_table_name, service_binary_dll_map, filter_func):
         kernel = self.context.modules[self.config["kernel"]]
 
-        service_table_name = self.create_service_table(
-            self.context, kernel.symbol_table_name, self.config_path
-        )
-
-        # Building the dictionary ahead of time is much better for performance
-        # vs looking up each service's DLL individually.
-        services_key = self._get_service_key(kernel)
-        service_binary_dll_map = (
-            self._get_service_binary_map(services_key)
-            if services_key is not None
-            else {}
-        )
-
         relative_tag_offset = self.context.symbol_space.get_type(
             service_table_name + constants.BANG + "_SERVICE_RECORD"
         ).relative_child_offset("Tag")
 
-        filter_func = pslist.PsList.create_name_filter(["services.exe"])
-
         is_vista_or_later = versions.is_vista_or_later(
             context=self.context, symbol_table=kernel.symbol_table_name
         )
@@ -306,37 +322,42 @@ class SvcScan(interfaces.plugins.PluginInterface):
                             renderers.UnreadableValue(), renderers.UnreadableValue()
                         ),
                     )
-                    yield (
-                        0,
-                        self.get_record_tuple(service_record, service_info),
-                    )
+                    yield self.get_record_tuple(service_record, service_info)
                 else:
-                    service_header = self.context.object(
-                        service_table_name + constants.BANG + "_SERVICE_HEADER",
-                        offset=offset,
-                        layer_name=proc_layer_name,
-                    )
-
-                    if not service_header.is_valid():
-                        continue
-
-                    # since we walk the s-list backwards, if we've seen
-                    # an object, then we've also seen all objects that
-                    # exist before it, thus we can break at that time.
-                    for service_record in service_header.ServiceRecord.traverse():
+                    for service_record in self.enumerate_vista_or_later_header(service_table_name, service_binary_dll_map, proc_layer_name, offset):
                         if service_record in seen:
                             break
                         seen.append(service_record)
-                        service_info = service_binary_dll_map.get(
-                            service_record.get_name(),
-                            ServiceBinaryInfo(
-                                renderers.UnreadableValue(), renderers.UnreadableValue()
-                            ),
-                        )
-                        yield (
-                            0,
-                            self.get_record_tuple(service_record, service_info),
-                        )
+                        yield service_record
+
+
+    def get_prereq_info(self):
+        """
+        Data structures and information needed to analyze service information
+        """
+        kernel = self.context.modules[self.config["kernel"]]
+
+        service_table_name = self.create_service_table(
+            self.context, kernel.symbol_table_name, self.config_path
+        )
+
+        services_key = self._get_service_key(kernel)
+
+        service_binary_dll_map = (
+            self._get_service_binary_map(services_key)
+            if services_key is not None
+            else {}
+        )
+
+        filter_func = pslist.PsList.create_name_filter(["services.exe"])
+
+        return service_table_name, service_binary_dll_map, filter_func
+
+    def _generator(self):
+        service_table_name, service_binary_dll_map, filter_func = self.get_prereq_info()
+
+        for record in self.service_scan(service_table_name, service_binary_dll_map, filter_func):
+            yield (0, record)
 
     def run(self):
         return renderers.TreeGrid(

volatility3/framework/symbols/windows/versions.py

@@ -141,6 +141,15 @@ is_win10_15063 = OsDistinguisher(
     ],
 )
 
+is_win10_15063_or_later = OsDistinguisher(
+    version_check=lambda x: x >= (10, 0, 15063),
+    fallback_checks=[
+        ("ObHeaderCookie", None, True),
+        ("_HANDLE_TABLE", "HandleCount", False),
+        ("_EPROCESS", "KeepAliveCounter", False),
+    ],
+)
+
 is_win10_16299_or_later = OsDistinguisher(
     version_check=lambda x: x >= (10, 0, 16299),
     fallback_checks=[