Malware categorization: windows.unhooked_system_calls

volatility3/framework/plugins/windows/malware/unhooked_system_calls.py

@@ -0,0 +1,202 @@
+# 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
+
+# Full details on the techniques used in these plugins to detect EDR-evading malware
+# can be found in our 20 page whitepaper submitted to DEFCON along with the presentation
+# https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf
+
+import logging
+
+from typing import Dict, Tuple, List, Generator
+
+from volatility3.framework import interfaces, exceptions
+from volatility3.framework import renderers
+from volatility3.framework.configuration import requirements
+from volatility3.framework.objects import utility
+from volatility3.plugins.windows import pslist, pe_symbols
+
+vollog = logging.getLogger(__name__)
+
+
+class UnhookedSystemCalls(interfaces.plugins.PluginInterface):
+    """Detects hooked ntdll.dll stub functions in Windows processes."""
+
+    _required_framework_version = (2, 4, 0)
+    _version = (2, 0, 0)
+
+    system_calls = {
+        "ntdll.dll": {
+            pe_symbols.wanted_names_identifier: [
+                "NtCreateThread",
+                "NtProtectVirtualMemory",
+                "NtReadVirtualMemory",
+                "NtOpenProcess",
+                "NtWriteFile",
+                "NtQueryVirtualMemory",
+                "NtAllocateVirtualMemory",
+                "NtWorkerFactoryWorkerReady",
+                "NtAcceptConnectPort",
+                "NtAddDriverEntry",
+                "NtAdjustPrivilegesToken",
+                "NtAlpcCreatePort",
+                "NtClose",
+                "NtCreateFile",
+                "NtCreateMutant",
+                "NtOpenFile",
+                "NtOpenIoCompletion",
+                "NtOpenJobObject",
+                "NtOpenKey",
+                "NtOpenKeyEx",
+                "NtOpenThread",
+                "NtOpenThreadToken",
+                "NtOpenThreadTokenEx",
+                "NtWriteVirtualMemory",
+                "NtTraceEvent",
+                "NtTranslateFilePath",
+                "NtUmsThreadYield",
+                "NtUnloadDriver",
+                "NtUnloadKey",
+                "NtUnloadKey2",
+                "NtUnloadKeyEx",
+                "NtCreateKey",
+                "NtCreateSection",
+                "NtDeleteKey",
+                "NtDeleteValueKey",
+                "NtDuplicateObject",
+                "NtQueryValueKey",
+                "NtReplaceKey",
+                "NtRequestWaitReplyPort",
+                "NtRestoreKey",
+                "NtSetContextThread",
+                "NtSetSecurityObject",
+                "NtSetValueKey",
+                "NtSystemDebugControl",
+                "NtTerminateProcess",
+            ]
+        }
+    }
+
+    # This data structure is used to track unique implementations of functions across processes
+    # The outer dictionary holds the module name (e.g., ntdll.dll)
+    # The next dictionary holds the function names (NtTerminateProcess, NtSetValueKey, etc.) inside a module
+    # The innermost dictionary holds the unique implementation (bytes) of a function across processes
+    # Each implementation is tracked along with the process(es) that host it
+    # For systems without malware, all functions should have the same implementation
+    # When API hooking/module unhooking is done, the victim (infected) processes will have unique implementations
+    _code_bytes_type = Dict[str, Dict[str, Dict[bytes, List[Tuple[int, str]]]]]
+
+    @classmethod
+    def get_requirements(cls) -> List:
+        # 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="pslist", component=pslist.PsList, version=(3, 0, 0)
+            ),
+            requirements.VersionRequirement(
+                name="pe_symbols", component=pe_symbols.PESymbols, version=(3, 0, 0)
+            ),
+        ]
+
+    def _gather_code_bytes(
+        self,
+        kernel_module_name: str,
+        found_symbols: pe_symbols.found_symbols_type,
+    ) -> _code_bytes_type:
+        """
+        Enumerates the desired DLLs and function implementations in each process
+        Groups based on unique implementations of each DLLs' functions
+        The purpose is to detect when a function has different implementations (code)
+        in different processes.
+        This very effectively detects code injection.
+        """
+        code_bytes: UnhookedSystemCalls._code_bytes_type = {}
+
+        procs = pslist.PsList.list_processes(self.context, kernel_module_name)
+
+        for proc in procs:
+            try:
+                proc_id = proc.UniqueProcessId
+                proc_name = utility.array_to_string(proc.ImageFileName)
+                proc_layer_name = proc.add_process_layer()
+            except exceptions.InvalidAddressException:
+                continue
+
+            for dll_name, functions in found_symbols.items():
+                for func_name, func_addr in functions:
+                    try:
+                        fbytes = self.context.layers[proc_layer_name].read(
+                            func_addr, 0x20
+                        )
+                    except exceptions.InvalidAddressException:
+                        continue
+
+                    # see the definition of _code_bytes_type for details of this data structure
+                    if dll_name not in code_bytes:
+                        code_bytes[dll_name] = {}
+
+                    if func_name not in code_bytes[dll_name]:
+                        code_bytes[dll_name][func_name] = {}
+
+                    if fbytes not in code_bytes[dll_name][func_name]:
+                        code_bytes[dll_name][func_name][fbytes] = []
+
+                    code_bytes[dll_name][func_name][fbytes].append((proc_id, proc_name))
+
+        return code_bytes
+
+    def _generator(self) -> Generator[Tuple[int, Tuple[str, str, int]], None, None]:
+        found_symbols = pe_symbols.PESymbols.addresses_for_process_symbols(
+            context=self.context,
+            config_path=self.config_path,
+            kernel_module_name=self.config["kernel"],
+            symbols=UnhookedSystemCalls.system_calls,
+        )
+
+        # code_bytes[dll_name][func_name][func_bytes]
+        code_bytes = self._gather_code_bytes(self.config["kernel"], found_symbols)
+
+        # walk the functions that were evaluated
+        for functions in code_bytes.values():
+            # cbb is the distinct groups of bytes (instructions)
+            # for this function across processes
+            for func_name, cbb in functions.items():
+                # the dict key here is the raw instructions, which is not helpful to look at
+                # the values are the list of tuples for the (proc_id, proc_name) pairs for this set of bytes (instructions)
+                cb = list(cbb.values())
+
+                # if all processes map to the same implementation, then no malware is present
+                if len(cb) == 1:
+                    yield 0, (func_name, "", len(cb[0]))
+                else:
+                    # if there are differing implementations then it means
+                    # that malware has overwritten system call(s) in infected processes
+                    # max_idx and small_idx find which implementation of a system call has the least processes
+                    # as all observed malware and open source projects only infected a few targets, leaving the
+                    # rest with the original EDR hooks in place
+                    max_idx = 0 if len(cb[0]) > len(cb[1]) else 1
+                    small_idx = (~max_idx) & 1
+
+                    ps = []
+
+                    # gather processes on small_idx since these are the malware infected ones
+                    for pid, pname in cb[small_idx]:
+                        ps.append(f"{pid:d}:{pname}")
+
+                    proc_names = ", ".join(ps)
+
+                    yield 0, (func_name, proc_names, len(cb[max_idx]))
+
+    def run(self) -> renderers.TreeGrid:
+        return renderers.TreeGrid(
+            [
+                ("Function", str),
+                ("Distinct Implementations", str),
+                ("Total Implementations", int),
+            ],
+            self._generator(),
+        )