Windows: SvcScan Binary Info

volatility3/framework/interfaces/context.py

@@ -87,7 +87,7 @@ def object(
         offset: int,
         native_layer_name: str = None,
         **arguments,
-    ):
+    ) -> "interfaces.objects.ObjectInterface":
         """Object factory, takes a context, symbol, offset and optional
         layer_name.
 

volatility3/framework/plugins/windows/svcscan.py

@@ -4,25 +4,42 @@
 
 import logging
 import os
-from typing import List
+from typing import Dict, List, NamedTuple, Optional, Tuple, Union, cast
 
-from volatility3.framework import interfaces, renderers, constants, symbols, exceptions
+from volatility3.framework import (
+    constants,
+    exceptions,
+    interfaces,
+    objects,
+    renderers,
+    symbols,
+)
 from volatility3.framework.configuration import requirements
 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.plugins.windows import poolscanner, vadyarascan, pslist
+from volatility3.plugins.windows import poolscanner, pslist, vadyarascan
+from volatility3.plugins.windows.registry import hivelist
 
 vollog = logging.getLogger(__name__)
 
 
+ServiceBinaryInfo = NamedTuple(
+    "ServiceBinaryInfo",
+    [
+        ("dll", Union[str, interfaces.renderers.BaseAbsentValue]),
+        ("binary", Union[str, interfaces.renderers.BaseAbsentValue]),
+    ],
+)
+
+
 class SvcScan(interfaces.plugins.PluginInterface):
     """Scans for windows services."""
 
     _required_framework_version = (2, 0, 0)
-    _version = (1, 0, 0)
+    _version = (2, 0, 0)
 
     @classmethod
     def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -42,10 +59,16 @@ def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]
             requirements.PluginRequirement(
                 name="vadyarascan", plugin=vadyarascan.VadYaraScan, version=(1, 0, 0)
             ),
+            requirements.PluginRequirement(
+                name="hivelist", plugin=hivelist.HiveList, version=(1, 0, 0)
+            ),
         ]
 
     @staticmethod
-    def get_record_tuple(service_record: interfaces.objects.ObjectInterface):
+    def get_record_tuple(
+        service_record: interfaces.objects.ObjectInterface,
+        binary_info: ServiceBinaryInfo,
+    ):
         return (
             format_hints.Hex(service_record.vol.offset),
             service_record.Order,
@@ -56,8 +79,32 @@ def get_record_tuple(service_record: interfaces.objects.ObjectInterface):
             service_record.get_name(),
             service_record.get_display(),
             service_record.get_binary(),
+            binary_info.binary,
+            binary_info.dll,
         )
 
+    # These checks must be completed from newest -> oldest OS version.
+    _win_version_file_map: List[Tuple[versions.OsDistinguisher, bool, str]] = [
+        (versions.is_win10_25398_or_later, True, "services-win10-25398-x64"),
+        (versions.is_win10_19041_or_later, True, "services-win10-19041-x64"),
+        (versions.is_win10_19041_or_later, False, "services-win10-19041-x86"),
+        (versions.is_win10_18362_or_later, True, "services-win10-18362-x64"),
+        (versions.is_win10_18362_or_later, False, "services-win10-18362-x86"),
+        (versions.is_win10_17763_or_later, False, "services-win10-17763-x86"),
+        (versions.is_win10_16299_or_later, True, "services-win10-16299-x64"),
+        (versions.is_win10_16299_or_later, False, "services-win10-16299-x86"),
+        (versions.is_win10_15063, True, "services-win10-15063-x64"),
+        (versions.is_win10_15063, False, "services-win10-15063-x86"),
+        (versions.is_win10_up_to_15063, True, "services-win8-x64"),
+        (versions.is_win10_up_to_15063, False, "services-win8-x86"),
+        (versions.is_windows_8_or_later, True, "services-win8-x64"),
+        (versions.is_windows_8_or_later, True, "services-win8-x86"),
+        (versions.is_vista_or_later, True, "services-vista-x64"),
+        (versions.is_vista_or_later, False, "services-vista-x86"),
+        (versions.is_windows_xp, False, "services-xp-x86"),
+        (versions.is_xp_or_2003, True, "services-xp-2003-x64"),
+    ]
+
     @staticmethod
     def create_service_table(
         context: interfaces.context.ContextInterface,
@@ -78,67 +125,14 @@ def create_service_table(
         native_types = context.symbol_space[symbol_table].natives
         is_64bit = symbols.symbol_table_is_64bit(context, symbol_table)
 
-        if (
-            versions.is_windows_xp(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-xp-x86"
-        elif (
-            versions.is_xp_or_2003(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-xp-2003-x64"
-        elif (
-            versions.is_win10_16299_or_later(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-win10-16299-x64"
-        elif (
-            versions.is_win10_16299_or_later(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-win10-16299-x86"
-        elif (
-            versions.is_win10_up_to_15063(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-win8-x64"
-        elif (
-            versions.is_win10_up_to_15063(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-win8-x86"
-        elif (
-            versions.is_win10_15063(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-win10-15063-x64"
-        elif (
-            versions.is_win10_15063(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-win10-15063-x86"
-        elif (
-            versions.is_windows_8_or_later(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-win8-x64"
-        elif (
-            versions.is_windows_8_or_later(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-win8-x86"
-        elif (
-            versions.is_vista_or_later(context=context, symbol_table=symbol_table)
-            and is_64bit
-        ):
-            symbol_filename = "services-vista-x64"
-        elif (
-            versions.is_vista_or_later(context=context, symbol_table=symbol_table)
-            and not is_64bit
-        ):
-            symbol_filename = "services-vista-x86"
-        else:
+        try:
+            symbol_filename = next(
+                filename
+                for version_check, for_64bit, filename in SvcScan._win_version_file_map
+                if is_64bit == for_64bit
+                and version_check(context=context, symbol_table=symbol_table)
+            )
+        except StopIteration:
             raise NotImplementedError("This version of Windows is not supported!")
 
         return intermed.IntermediateSymbolTable.create(
@@ -150,13 +144,110 @@ def create_service_table(
             native_types=native_types,
         )
 
+    def _get_service_key(self, kernel) -> Optional[objects.StructType]:
+        for hive in hivelist.HiveList.list_hives(
+            context=self.context,
+            base_config_path=interfaces.configuration.path_join(
+                self.config_path, "hivelist"
+            ),
+            layer_name=kernel.layer_name,
+            symbol_table=kernel.symbol_table_name,
+            filter_string="machine\\system",
+        ):
+            # Get ControlSet\Services.
+            try:
+                return cast(
+                    objects.StructType, hive.get_key(r"CurrentControlSet\Services")
+                )
+            except (KeyError, exceptions.InvalidAddressException):
+                try:
+                    return cast(
+                        objects.StructType, hive.get_key(r"ControlSet001\Services")
+                    )
+                except (KeyError, exceptions.InvalidAddressException):
+                    vollog.log(
+                        constants.LOGLEVEL_VVVV,
+                        "Could not retrieve any control set from SYSTEM hive",
+                    )
+
+        return None
+
+    @staticmethod
+    def _get_service_dll(
+        service_key,
+    ) -> Union[str, interfaces.renderers.BaseAbsentValue]:
+        try:
+            param_key = next(
+                key
+                for key in service_key.get_subkeys()
+                if key.get_name() == "Parameters"
+            )
+            return (
+                next(
+                    val
+                    for val in param_key.get_values()
+                    if val.get_name() == "ServiceDll"
+                )
+                .decode_data()
+                .decode("utf-16")
+                .rstrip("\x00")
+            )
+
+        except UnicodeDecodeError:
+            return renderers.UnparsableValue()
+        except StopIteration:
+            return renderers.UnreadableValue()
+
+    @staticmethod
+    def _get_service_binary(
+        service_key,
+    ) -> Union[str, interfaces.renderers.BaseAbsentValue]:
+        try:
+            return (
+                next(
+                    val
+                    for val in service_key.get_values()
+                    if val.get_name() == "ImagePath"
+                )
+                .decode_data()
+                .decode("utf-16")
+                .rstrip("\x00")
+            )
+
+        except UnicodeDecodeError:
+            return renderers.UnparsableValue()
+        except StopIteration:
+            return renderers.UnreadableValue()
+
+    @staticmethod
+    def _get_service_binary_map(
+        services_key: interfaces.objects.ObjectInterface,
+    ) -> Dict[str, ServiceBinaryInfo]:
+        services = services_key.get_subkeys()
+        return {
+            service_key.get_name(): ServiceBinaryInfo(
+                SvcScan._get_service_dll(service_key),
+                SvcScan._get_service_binary(service_key),
+            )
+            for service_key in services
+        }
+
     def _generator(self):
         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")
@@ -209,7 +300,16 @@ def _generator(self):
                     if not service_record.is_valid():
                         continue
 
-                    yield (0, self.get_record_tuple(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),
+                    )
                 else:
                     service_header = self.context.object(
                         service_table_name + constants.BANG + "_SERVICE_HEADER",
@@ -227,7 +327,16 @@ def _generator(self):
                         if service_record in seen:
                             break
                         seen.append(service_record)
-                        yield (0, self.get_record_tuple(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),
+                        )
 
     def run(self):
         return renderers.TreeGrid(
@@ -241,6 +350,8 @@ def run(self):
                 ("Name", str),
                 ("Display", str),
                 ("Binary", str),
+                ("Binary (Registry)", str),
+                ("Dll", str),
             ],
             self._generator(),
         )