mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-28 12:34:53 +02:00
Merge pull request #1250 from volatilityfoundation/pe_symbols_new
Add new pe_symbols API, debug registers plugin, unhooked system calls…
This commit is contained in:
ikelos
@@ -0,0 +1,211 @@
|
||||
# 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 Tuple, Optional, Generator, List, Dict
|
||||
|
||||
from functools import partial
|
||||
|
||||
from volatility3.framework import renderers, interfaces, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.framework.renderers import format_hints
|
||||
import volatility3.plugins.windows.pslist as pslist
|
||||
import volatility3.plugins.windows.threads as threads
|
||||
import volatility3.plugins.windows.pe_symbols as pe_symbols
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DebugRegisters(interfaces.plugins.PluginInterface):
|
||||
# version 2.6.0 adds support for scanning for 'Ethread' structures by pool tags
|
||||
_required_framework_version = (2, 6, 0)
|
||||
_version = (1, 0, 0)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List:
|
||||
return [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _get_debug_info(
|
||||
ethread: interfaces.objects.ObjectInterface,
|
||||
) -> Optional[Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]]:
|
||||
"""
|
||||
Gathers information related to the debug registers for the given thread
|
||||
Args:
|
||||
ethread: the thread (_ETHREAD) to examine
|
||||
Returns:
|
||||
Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]: The owner process of the thread and the values for dr7, dr0, dr1, dr2, dr3
|
||||
"""
|
||||
try:
|
||||
dr7 = ethread.Tcb.TrapFrame.Dr7
|
||||
state = ethread.Tcb.State
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# 0 = debug registers not active
|
||||
# 4 = terminated
|
||||
if dr7 == 0 or state == 4:
|
||||
return None
|
||||
|
||||
try:
|
||||
owner_proc = ethread.owning_process()
|
||||
except (AttributeError, exceptions.InvalidAddressException):
|
||||
return None
|
||||
|
||||
dr0 = ethread.Tcb.TrapFrame.Dr0
|
||||
dr1 = ethread.Tcb.TrapFrame.Dr1
|
||||
dr2 = ethread.Tcb.TrapFrame.Dr2
|
||||
dr3 = ethread.Tcb.TrapFrame.Dr3
|
||||
|
||||
# bail if all are 0
|
||||
if not (dr0 or dr1 or dr2 or dr3):
|
||||
return None
|
||||
|
||||
return owner_proc, dr7, dr0, dr1, dr2, dr3
|
||||
|
||||
def _generator(
|
||||
self,
|
||||
) -> Generator[
|
||||
Tuple[
|
||||
int,
|
||||
Tuple[
|
||||
str,
|
||||
int,
|
||||
int,
|
||||
int,
|
||||
int,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
],
|
||||
],
|
||||
None,
|
||||
None,
|
||||
]:
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
|
||||
vads_cache: Dict[int, pe_symbols.ranges_type] = {}
|
||||
|
||||
proc_modules = None
|
||||
|
||||
procs = pslist.PsList.list_processes(
|
||||
context=self.context,
|
||||
layer_name=kernel.layer_name,
|
||||
symbol_table=kernel.symbol_table_name,
|
||||
)
|
||||
|
||||
for proc in procs:
|
||||
for thread in threads.Threads.list_threads(kernel, proc):
|
||||
debug_info = self._get_debug_info(thread)
|
||||
if not debug_info:
|
||||
continue
|
||||
|
||||
owner_proc, dr7, dr0, dr1, dr2, dr3 = debug_info
|
||||
|
||||
vads = pe_symbols.PESymbols.get_vads_for_process_cache(
|
||||
vads_cache, owner_proc
|
||||
)
|
||||
if not vads:
|
||||
continue
|
||||
|
||||
# this lookup takes a while, so only perform if we need to
|
||||
if not proc_modules:
|
||||
proc_modules = pe_symbols.PESymbols.get_process_modules(
|
||||
self.context, kernel.layer_name, kernel.symbol_table_name, None
|
||||
)
|
||||
path_and_symbol = partial(
|
||||
pe_symbols.PESymbols.path_and_symbol_for_address,
|
||||
self.context,
|
||||
self.config_path,
|
||||
proc_modules,
|
||||
)
|
||||
|
||||
file0, sym0 = path_and_symbol(vads, dr0)
|
||||
file1, sym1 = path_and_symbol(vads, dr1)
|
||||
file2, sym2 = path_and_symbol(vads, dr2)
|
||||
file3, sym3 = path_and_symbol(vads, dr3)
|
||||
|
||||
# if none map to an actual file VAD then bail
|
||||
if not (file0 or file1 or file2 or file3):
|
||||
continue
|
||||
|
||||
process_name = owner_proc.ImageFileName.cast(
|
||||
"string",
|
||||
max_length=owner_proc.ImageFileName.vol.count,
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
thread_tid = thread.Cid.UniqueThread
|
||||
|
||||
yield (
|
||||
0,
|
||||
(
|
||||
process_name,
|
||||
owner_proc.UniqueProcessId,
|
||||
thread_tid,
|
||||
thread.Tcb.State,
|
||||
dr7,
|
||||
format_hints.Hex(dr0),
|
||||
file0 or renderers.NotApplicableValue(),
|
||||
sym0 or renderers.NotApplicableValue(),
|
||||
format_hints.Hex(dr1),
|
||||
file1 or renderers.NotApplicableValue(),
|
||||
sym1 or renderers.NotApplicableValue(),
|
||||
format_hints.Hex(dr2),
|
||||
file2 or renderers.NotApplicableValue(),
|
||||
sym2 or renderers.NotApplicableValue(),
|
||||
format_hints.Hex(dr3),
|
||||
file3 or renderers.NotApplicableValue(),
|
||||
sym3 or renderers.NotApplicableValue(),
|
||||
),
|
||||
)
|
||||
|
||||
def run(self) -> renderers.TreeGrid:
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
("Process", str),
|
||||
("PID", int),
|
||||
("TID", int),
|
||||
("State", int),
|
||||
("Dr7", int),
|
||||
("Dr0", format_hints.Hex),
|
||||
("Range0", str),
|
||||
("Symbol0", str),
|
||||
("Dr1", format_hints.Hex),
|
||||
("Range1", str),
|
||||
("Symbol1", str),
|
||||
("Dr2", format_hints.Hex),
|
||||
("Range2", str),
|
||||
("Symbol2", str),
|
||||
("Dr3", format_hints.Hex),
|
||||
("Range3", str),
|
||||
("Symbol3", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,208 @@
|
||||
# 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 unhooked_system_calls(interfaces.plugins.PluginInterface):
|
||||
"""Looks for signs of Skeleton Key malware"""
|
||||
|
||||
_required_framework_version = (2, 4, 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=(2, 0, 0)
|
||||
),
|
||||
requirements.PluginRequirement(
|
||||
name="pe_symbols", plugin=pe_symbols.PESymbols, version=(1, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
def _gather_code_bytes(
|
||||
self,
|
||||
kernel: interfaces.context.ModuleInterface,
|
||||
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: unhooked_system_calls._code_bytes_type = {}
|
||||
|
||||
procs = pslist.PsList.list_processes(
|
||||
context=self.context,
|
||||
layer_name=kernel.layer_name,
|
||||
symbol_table=kernel.symbol_table_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]:
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
|
||||
found_symbols = pe_symbols.PESymbols.addresses_for_process_symbols(
|
||||
self.context,
|
||||
self.config_path,
|
||||
kernel.layer_name,
|
||||
kernel.symbol_table_name,
|
||||
unhooked_system_calls.system_calls,
|
||||
)
|
||||
|
||||
# code_bytes[dll_name][func_name][func_bytes]
|
||||
code_bytes = self._gather_code_bytes(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("{:d}:{}".format(pid, 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(),
|
||||
)
|
||||
@@ -3,7 +3,7 @@
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import Callable, List, Generator, Iterable, Type, Optional
|
||||
from typing import Callable, List, Generator, Iterable, Type, Optional, Tuple
|
||||
|
||||
from volatility3.framework import renderers, interfaces, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
@@ -196,11 +196,31 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
|
||||
return file_handle
|
||||
|
||||
def _generator(self, procs):
|
||||
def _generator(self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[
|
||||
Tuple[
|
||||
int,
|
||||
Tuple[
|
||||
int,
|
||||
str,
|
||||
format_hints.Hex,
|
||||
format_hints.Hex,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
int,
|
||||
int,
|
||||
format_hints.Hex,
|
||||
str,
|
||||
str,
|
||||
],
|
||||
],
|
||||
None,
|
||||
None,
|
||||
]:
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
kernel_layer = self.context.layers[kernel.layer_name]
|
||||
|
||||
def passthrough(_: interfaces.objects.ObjectInterface) -> bool:
|
||||
def passthrough(x: interfaces.objects.ObjectInterface) -> bool:
|
||||
return False
|
||||
|
||||
filter_func = passthrough
|
||||
@@ -250,7 +270,7 @@ class VadInfo(interfaces.plugins.PluginInterface):
|
||||
),
|
||||
)
|
||||
|
||||
def run(self):
|
||||
def run(self) -> renderers.TreeGrid:
|
||||
kernel = self.context.modules[self.config["kernel"]]
|
||||
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
|
||||
|
||||
Reference in New Issue
Block a user