mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-17 20:35:40 +02:00
Merge pull request #1852 from SolitudePy/syscalls_malware_plugin
Malware categorization: categorize EDR evading plugins with in/direct_syscall
This commit is contained in:
ikelos
@@ -1,28 +1,13 @@
|
||||
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# This file is Copyright 2025 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 volatility3.framework import interfaces, deprecation
|
||||
from collections import namedtuple
|
||||
from typing import List, Tuple, Optional, Generator, Callable
|
||||
|
||||
from volatility3.framework.objects import utility
|
||||
from volatility3.framework import interfaces, renderers, symbols, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.plugins import yarascan
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.plugins.windows import pslist
|
||||
from volatility3.plugins.windows.malware import direct_system_calls
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
import capstone
|
||||
|
||||
has_capstone = True
|
||||
except ImportError:
|
||||
has_capstone = False
|
||||
|
||||
# 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
|
||||
@@ -38,19 +23,14 @@ syscall_finder_type = namedtuple(
|
||||
],
|
||||
)
|
||||
|
||||
syscall_finder_type.__doc__ = """
|
||||
This type is used to specify how malicious system call invocations should be found.
|
||||
|
||||
`get_syscall_target_address` is optionally used to extract the address containing the malicious 'syscall' instruction
|
||||
`wants_syscall_inst` whether or not this method expects the 'syscall' instruction directly within the malicious code block
|
||||
`rule` the opcode string to search for the malicious syscall instructions
|
||||
`invalid_ops` instructions that only appear in invalid code blocks. Stops processing of the code block when encountered.
|
||||
`termination_ops` instructions that are expected to be present in the code block and that stop processing
|
||||
"""
|
||||
|
||||
|
||||
class DirectSystemCalls(interfaces.plugins.PluginInterface):
|
||||
"""Detects the Direct System Call technique used to bypass EDRs"""
|
||||
class DirectSystemCalls(
|
||||
interfaces.plugins.PluginInterface,
|
||||
deprecation.PluginRenameClass,
|
||||
replacement_class=direct_system_calls.DirectSystemCalls,
|
||||
removal_date="2026-06-07",
|
||||
):
|
||||
"""Detects the Direct System Call technique used to bypass EDRs (deprecated)."""
|
||||
|
||||
_required_framework_version = (2, 4, 0)
|
||||
|
||||
@@ -81,392 +61,3 @@ class DirectSystemCalls(interfaces.plugins.PluginInterface):
|
||||
# the expected form is to end with a "ret" back to the calling code
|
||||
["ret"],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
# create a list of requirements for vadyarascan
|
||||
vadyarascan_requirements = [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(3, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascanner", component=yarascan.YaraScanner, version=(2, 1, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascan", component=yarascan.YaraScan, version=(2, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
# get base yarascan requirements for command line options
|
||||
yarascan_requirements = yarascan.YaraScan.get_yarascan_option_requirements()
|
||||
|
||||
# return the combined requirements
|
||||
return yarascan_requirements + vadyarascan_requirements
|
||||
|
||||
@staticmethod
|
||||
def _is_syscall_block(
|
||||
disasm_func: Callable,
|
||||
syscall_finder: syscall_finder_type,
|
||||
data: bytes,
|
||||
address: int,
|
||||
) -> Optional[Tuple[str, "capstone._cs_insn"]]:
|
||||
"""
|
||||
Determines if the bytes starting at `data` represent a valid syscall instruction invocation block
|
||||
|
||||
To maliciously invoke the system call instruction, malware must do each of the following:
|
||||
|
||||
1) update RAX to the system call number
|
||||
2) update R10 to the first parameter
|
||||
3) hit the 'termination' instruction set in `syscall_finder_type`
|
||||
|
||||
We also track whether the 'syscall' instruction was encountered while parsing
|
||||
|
||||
This function is reusable for every technique we found and studied during the DEFCON research timeframe
|
||||
|
||||
Args:
|
||||
disasm_func: capstone disassembly function gathered from `get_disasm_function`
|
||||
syscall_finder: the method and constraints on the malicious system call blocks that the calling plugin knows how to find
|
||||
data: the bytes from memory to search for malicious syscall invocations
|
||||
address: the address from where `data` came from in the particular process
|
||||
Returns:
|
||||
Optional[Tuple[str, capstone._cs_insn]]: For valid blocks, the disassembled bytes in string from and the last (termination) instruction
|
||||
"""
|
||||
found_movr10 = False
|
||||
found_movreax = False
|
||||
found_syscall = False
|
||||
found_end = False
|
||||
end_inst = None
|
||||
|
||||
disasm_bytes = ""
|
||||
|
||||
for inst in disasm_func(data, address):
|
||||
disasm_bytes += f"{inst.address:#x}: {inst.mnemonic} {inst.op_str}; "
|
||||
|
||||
# an instruction of all 0x00 opcodes
|
||||
if inst.opcode.count(0) == len(inst.opcode):
|
||||
break
|
||||
|
||||
op = inst.mnemonic
|
||||
|
||||
# invalid op, bail
|
||||
if op in syscall_finder.invalid_ops:
|
||||
break
|
||||
|
||||
# found the end instruction wanted by the caller
|
||||
elif op in syscall_finder.termination_ops:
|
||||
found_end = True
|
||||
end_inst = inst
|
||||
break
|
||||
|
||||
# track this no matter what to make code more re-usable
|
||||
elif op == "syscall":
|
||||
found_syscall = True
|
||||
|
||||
# if we hit a 'syscall' but RAX or R10 haven't been touched
|
||||
# then we are in an invalid path, so bail
|
||||
if not syscall_finder.wants_syscall_inst or (
|
||||
not (found_movr10 and found_movreax)
|
||||
):
|
||||
break
|
||||
|
||||
else:
|
||||
# attempt to see if any other instruction type wrote to registers
|
||||
try:
|
||||
_, regs_written = inst.regs_access()
|
||||
except capstone.CsError:
|
||||
continue
|
||||
|
||||
if regs_written:
|
||||
for r in regs_written:
|
||||
# track writes to eax/rax or R10
|
||||
reg = inst.reg_name(r)
|
||||
if reg in ["eax", "rax"]:
|
||||
found_movreax = True
|
||||
|
||||
elif reg == "r10":
|
||||
found_movr10 = True
|
||||
|
||||
# if any of these are missing, the block is invalid regardless of
|
||||
# the technique we are trying to detect now or in the future
|
||||
if not (found_movr10 and found_movreax and found_end):
|
||||
return None
|
||||
|
||||
# if the finder requires a 'syscall' instruction then bail now if we didn't find one
|
||||
if syscall_finder.wants_syscall_inst and not found_syscall:
|
||||
return None
|
||||
|
||||
return disasm_bytes, end_inst
|
||||
|
||||
@classmethod
|
||||
def get_disasm_function(cls, architecture: str) -> Callable:
|
||||
"""
|
||||
Returns the disassembly handler for the given architecture
|
||||
.detail is used to get full instruction information
|
||||
|
||||
Args:
|
||||
architecture: the name of the architecture for the process being disassembled
|
||||
Returns:
|
||||
The disasm function from capstone for the given architecture
|
||||
"""
|
||||
disasm_types = {
|
||||
"intel": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32),
|
||||
"intel64": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64),
|
||||
}
|
||||
|
||||
disasm_type = disasm_types[architecture]
|
||||
disasm_type.detail = True
|
||||
return disasm_type.disasm
|
||||
|
||||
@classmethod
|
||||
def _is_valid_syscall(
|
||||
cls,
|
||||
syscall_finder: syscall_finder_type,
|
||||
proc_layer: interfaces.layers.DataLayerInterface,
|
||||
architecture: str,
|
||||
vads: List[Tuple[int, int, str]],
|
||||
address: int,
|
||||
) -> Optional[Tuple[int, str]]:
|
||||
"""
|
||||
Args:
|
||||
syscall_finder:
|
||||
proc_layer: the memory layer of the process being scanned
|
||||
architecture: the name of the architecture for the process being disassembled
|
||||
vads: the ranges of this process under 10MB
|
||||
address: the starting address to check for malicious syscall code blocks
|
||||
|
||||
Returns:
|
||||
Optional[Tuple[int, str]]: For valid code blocks, the starting address of the block and the disassembly string
|
||||
"""
|
||||
# the number bytes behind the yara rule hit to scan
|
||||
behind = 32
|
||||
|
||||
address = address - behind
|
||||
|
||||
try:
|
||||
data = proc_layer.read(address, behind * 2)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
disasm_func = cls.get_disasm_function(architecture)
|
||||
|
||||
# since Intel does not have fixed-size instructions, we have to scan
|
||||
# each byte offset and re-disassemble the remaining block
|
||||
for offset in range(behind):
|
||||
# if this looks like a system call back (r10, rax, ret/jmp)
|
||||
syscall_info = cls._is_syscall_block(
|
||||
disasm_func, syscall_finder, data[offset:], address + offset
|
||||
)
|
||||
if syscall_info:
|
||||
disasm_bytes, end_inst = syscall_info
|
||||
|
||||
# if we can recover (and require) a target address for this malware technique
|
||||
if syscall_finder.get_syscall_target_address:
|
||||
target_address = syscall_finder.get_syscall_target_address(
|
||||
proc_layer, end_inst
|
||||
)
|
||||
|
||||
# could not determine the address -> invalid basic block
|
||||
if not target_address:
|
||||
continue
|
||||
|
||||
# we only care about calls to system call DLLs
|
||||
path = cls.get_range_path(vads, target_address)
|
||||
if not isinstance(path, str) or not path.lower().endswith(
|
||||
cls.valid_syscall_handlers
|
||||
):
|
||||
continue
|
||||
|
||||
# return the address and disassembly string if all checks pass
|
||||
return address + offset, disasm_bytes
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_vad_maps(
|
||||
cls,
|
||||
task: interfaces.objects.ObjectInterface,
|
||||
) -> List[Tuple[int, int, str]]:
|
||||
"""Creates a map of start/end addresses within a virtual address
|
||||
descriptor tree.
|
||||
|
||||
Args:
|
||||
task: The EPROCESS object of which to traverse the vad tree
|
||||
|
||||
Returns:
|
||||
An iterable of tuples containing start and end addresses for each descriptor
|
||||
"""
|
||||
vads: List[Tuple[int, int, str]] = []
|
||||
|
||||
# scan regions under 10MB
|
||||
scan_max = 10 * 1000 * 1000
|
||||
|
||||
vad_root = task.get_vad_root()
|
||||
|
||||
for vad in vad_root.traverse():
|
||||
if vad.get_size() < scan_max:
|
||||
vads.append((vad.get_start(), vad.get_size(), vad.get_file_name()))
|
||||
|
||||
return vads
|
||||
|
||||
@classmethod
|
||||
def get_range_path(
|
||||
cls, ranges: List[Tuple[int, int, str]], address: int
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Returns the path for the range holding `address`, if found
|
||||
|
||||
Args:
|
||||
ranges: VADs collected from `get_vad_maps`
|
||||
address: the address to find
|
||||
Returns:
|
||||
The path holding the address, if any
|
||||
"""
|
||||
for start, size, path in ranges:
|
||||
if start <= address < start + size:
|
||||
return path
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_tasks_to_scan(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
kernel_module_name: str,
|
||||
) -> Generator[
|
||||
Tuple[interfaces.objects.ObjectInterface, str, str, str], None, None
|
||||
]:
|
||||
"""
|
||||
Gathers active processes with the extra information needed
|
||||
to detect malicious syscall instructions
|
||||
|
||||
Returns:
|
||||
Generator of the process object, name, memory layer, and architecture
|
||||
"""
|
||||
|
||||
# gather active processes
|
||||
filter_func = pslist.PsList.create_active_process_filter()
|
||||
|
||||
kernel = context.modules[kernel_module_name]
|
||||
|
||||
is_32bit_arch = not symbols.symbol_table_is_64bit(
|
||||
context=context, symbol_table_name=kernel.symbol_table_name
|
||||
)
|
||||
|
||||
for proc in pslist.PsList.list_processes(
|
||||
context=context,
|
||||
kernel_module_name=kernel_module_name,
|
||||
filter_func=filter_func,
|
||||
):
|
||||
proc_name = utility.array_to_string(proc.ImageFileName)
|
||||
|
||||
# skip Defender
|
||||
if proc_name in ["MsMpEng.exe"]:
|
||||
continue
|
||||
|
||||
try:
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
except exceptions.InvalidAddressException:
|
||||
continue
|
||||
|
||||
if is_32bit_arch or proc.get_is_wow64():
|
||||
architecture = "intel"
|
||||
else:
|
||||
architecture = "intel64"
|
||||
|
||||
yield proc, proc_name, proc_layer_name, architecture
|
||||
|
||||
@classmethod
|
||||
def _get_rule_hits(
|
||||
cls,
|
||||
context: interfaces.objects.ObjectInterface,
|
||||
proc_layer: interfaces.layers.DataLayerInterface,
|
||||
vads: List[Tuple[int, int, str]],
|
||||
pattern: str,
|
||||
) -> Generator[Tuple[int, Optional[str]], None, None]:
|
||||
"""
|
||||
Runs the given opcode rule through Yara and returns the address and file path of hits
|
||||
|
||||
Args:
|
||||
context:
|
||||
proc_layer: the layer to scan
|
||||
vads: the ranges inside of the process being scanned
|
||||
pattern: the opcodes rule from the plugin to detect a particular EDR-bypass technique
|
||||
|
||||
Returns:
|
||||
Generator of the address and file path of hits
|
||||
"""
|
||||
sections = [(vad[0], vad[1]) for vad in vads]
|
||||
|
||||
rule = yarascan.YaraScanner.get_rule(pattern)
|
||||
|
||||
for hit in proc_layer.scan(
|
||||
context=context,
|
||||
scanner=yarascan.YaraScanner(rules=rule),
|
||||
sections=sections,
|
||||
):
|
||||
address = hit[0]
|
||||
|
||||
path = cls.get_range_path(vads, address)
|
||||
|
||||
# ignore hits in the system call DLLs
|
||||
if isinstance(path, str) and path.lower().endswith(
|
||||
cls.valid_syscall_handlers
|
||||
):
|
||||
continue
|
||||
|
||||
yield address, path
|
||||
|
||||
def _generator(
|
||||
self,
|
||||
) -> Generator[Tuple[int, Tuple[str, int, Optional[str], int, str]], None, None]:
|
||||
if not has_capstone:
|
||||
vollog.warning(
|
||||
"capstone is not installed. This plugin requires capstone to operate."
|
||||
)
|
||||
return
|
||||
|
||||
for proc, proc_name, proc_layer_name, architecture in self.get_tasks_to_scan(
|
||||
self.context, self.config["kernel"]
|
||||
):
|
||||
proc_layer = self.context.layers[proc_layer_name]
|
||||
|
||||
vads = self.get_vad_maps(proc)
|
||||
if not vads:
|
||||
continue
|
||||
|
||||
# for each valid process, look for malicious syscall invocations
|
||||
for address, vad_path in self._get_rule_hits(
|
||||
self.context, proc_layer, vads, self.syscall_finder.rule_str
|
||||
):
|
||||
syscall_info = self._is_valid_syscall(
|
||||
self.syscall_finder, proc_layer, architecture, vads, address
|
||||
)
|
||||
if not syscall_info:
|
||||
continue
|
||||
|
||||
address, disasm_bytes = syscall_info
|
||||
|
||||
yield 0, (
|
||||
proc_name,
|
||||
proc.UniqueProcessId,
|
||||
vad_path,
|
||||
format_hints.Hex(address),
|
||||
disasm_bytes,
|
||||
)
|
||||
|
||||
def run(self) -> renderers.TreeGrid:
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
("Process", str),
|
||||
("PID", int),
|
||||
("Range", str),
|
||||
("Address", format_hints.Hex),
|
||||
("Disasm", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
|
||||
@@ -1,119 +1,21 @@
|
||||
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import struct
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
from volatility3.framework import interfaces, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.plugins import yarascan
|
||||
from volatility3.plugins.windows import direct_system_calls
|
||||
from volatility3.framework import deprecation
|
||||
from volatility3.plugins.windows.malware import indirect_system_calls
|
||||
from volatility3.plugins.windows.malware import direct_system_calls
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IndirectSystemCalls(direct_system_calls.DirectSystemCalls):
|
||||
class IndirectSystemCalls(
|
||||
direct_system_calls.DirectSystemCalls,
|
||||
deprecation.PluginRenameClass,
|
||||
replacement_class=indirect_system_calls.IndirectSystemCalls,
|
||||
removal_date="2026-06-07",
|
||||
):
|
||||
"""Detects the Indirect System Call technique used to bypass EDRs (deprecated)."""
|
||||
|
||||
_required_framework_version = (2, 4, 0)
|
||||
_version = (1, 0, 0)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
self.syscall_finder = direct_system_calls.syscall_finder_type(
|
||||
# gets the target address of a indirect jmp
|
||||
self._indirect_syscall_block_target,
|
||||
# we are looking for indirect system calls, so we don't want 'syscall' instructions in our code block
|
||||
False,
|
||||
# jmp [address]; ret
|
||||
"/\\xff\\x25[^\\xc3]{,24}\\xc3/",
|
||||
# any of these mean we aren't in a malicious indirect call
|
||||
["call", "leave", "int3", "ret"],
|
||||
# stop at jmp, this should reference the system call instruction
|
||||
["jmp"],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
# create a list of requirements for vadyarascan
|
||||
vadyarascan_requirements = [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascanner", component=yarascan.YaraScanner, version=(2, 1, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascan", component=yarascan.YaraScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="direct_system_calls",
|
||||
component=direct_system_calls.DirectSystemCalls,
|
||||
version=(2, 0, 0),
|
||||
),
|
||||
]
|
||||
|
||||
# get base yarascan requirements for command line options
|
||||
yarascan_requirements = yarascan.YaraScan.get_yarascan_option_requirements()
|
||||
|
||||
# return the combined requirements
|
||||
return yarascan_requirements + vadyarascan_requirements
|
||||
|
||||
@staticmethod
|
||||
def _indirect_syscall_block_target(
|
||||
proc_layer: interfaces.layers.DataLayerInterface, inst
|
||||
) -> Optional[int]:
|
||||
"""
|
||||
This function determines the address of a jmp in the following form:
|
||||
|
||||
jmp [address]
|
||||
|
||||
To determine this, we must:
|
||||
1) Pull the 4 byte relative offset of 'address' inside the instruction
|
||||
2) Compute the full address of this relative offset
|
||||
3) Read from the address as it is being dereferenced
|
||||
4) Ensure the target address points to a 'syscall' instruction
|
||||
|
||||
Args:
|
||||
proc_layer: the layer of the potential syscall block
|
||||
inst: the terminating instruction of the syscall block check
|
||||
Returns:
|
||||
The target address of the jump if it can be computed
|
||||
"""
|
||||
|
||||
try:
|
||||
jmp_address_str = proc_layer.read(inst.address, 6)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# Should be an jmp...
|
||||
if jmp_address_str[0:2] != b"\xff\x25":
|
||||
return None
|
||||
|
||||
# get the address of the 'jmp [address]' instruction
|
||||
relative_offset = struct.unpack("<I", jmp_address_str[2:])[0]
|
||||
if not relative_offset or relative_offset == -1:
|
||||
return None
|
||||
|
||||
# compute the target address of the jmp (dereference)
|
||||
jmp_address = inst.address + relative_offset + 6
|
||||
try:
|
||||
jmp_target_str = proc_layer.read(jmp_address, 8)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# compute from the target address then read from it
|
||||
jmp_target_address = struct.unpack("<Q", jmp_target_str)[0]
|
||||
try:
|
||||
jmp_target = proc_layer.read(jmp_target_address, 2)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# check that the address points to a 'syscall' instruction
|
||||
if jmp_target == b"\x0f\x05":
|
||||
return jmp_target_address
|
||||
|
||||
return None
|
||||
|
||||
@@ -0,0 +1,472 @@
|
||||
# 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
|
||||
#
|
||||
|
||||
import logging
|
||||
|
||||
from collections import namedtuple
|
||||
from typing import List, Tuple, Optional, Generator, Callable
|
||||
|
||||
from volatility3.framework.objects import utility
|
||||
from volatility3.framework import interfaces, renderers, symbols, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.plugins import yarascan
|
||||
from volatility3.framework.renderers import format_hints
|
||||
from volatility3.plugins.windows import pslist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
import capstone
|
||||
|
||||
has_capstone = True
|
||||
except ImportError:
|
||||
has_capstone = False
|
||||
|
||||
# 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
|
||||
|
||||
syscall_finder_type = namedtuple(
|
||||
"syscall_finder_type",
|
||||
[
|
||||
"get_syscall_target_address",
|
||||
"wants_syscall_inst",
|
||||
"rule_str",
|
||||
"invalid_ops",
|
||||
"termination_ops",
|
||||
],
|
||||
)
|
||||
|
||||
syscall_finder_type.__doc__ = """
|
||||
This type is used to specify how malicious system call invocations should be found.
|
||||
|
||||
`get_syscall_target_address` is optionally used to extract the address containing the malicious 'syscall' instruction
|
||||
`wants_syscall_inst` whether or not this method expects the 'syscall' instruction directly within the malicious code block
|
||||
`rule` the opcode string to search for the malicious syscall instructions
|
||||
`invalid_ops` instructions that only appear in invalid code blocks. Stops processing of the code block when encountered.
|
||||
`termination_ops` instructions that are expected to be present in the code block and that stop processing
|
||||
"""
|
||||
|
||||
|
||||
class DirectSystemCalls(interfaces.plugins.PluginInterface):
|
||||
"""Detects the Direct System Call technique used to bypass EDRs"""
|
||||
|
||||
_required_framework_version = (2, 4, 0)
|
||||
|
||||
# 2.0.0 - changes signature of `get_tasks_to_scan`
|
||||
_version = (2, 0, 0)
|
||||
|
||||
# DLLs that are expected to host system call invocations
|
||||
valid_syscall_handlers = ("ntdll.dll", "win32u.dll")
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
self.syscall_finder = syscall_finder_type(
|
||||
# for direct system calls, we find the `syscall` instruction directly, so we already know the address
|
||||
None,
|
||||
# yes, we want the syscall instruction present as it is what this technique looks for
|
||||
True,
|
||||
# regex to find "\x0f\x05" (syscall) followed later by "\xc3" (ret)
|
||||
# we allow spacing in between to break naive anti-analysis forms (e.g., TarTarus Gate)
|
||||
# Standard techniques, such as HellsGate, look like:
|
||||
# mov r10, rcx
|
||||
# mov eax, <system call number>
|
||||
# syscall
|
||||
# ret
|
||||
"/\\x0f\\x05[^\\xc3]{,24}\\xc3/",
|
||||
# any of these will not be in a workable, malicious direct system call block
|
||||
["jmp", "call", "leave", "int3"],
|
||||
# the expected form is to end with a "ret" back to the calling code
|
||||
["ret"],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
# create a list of requirements for vadyarascan
|
||||
vadyarascan_requirements = [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="pslist", component=pslist.PsList, version=(3, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascanner", component=yarascan.YaraScanner, version=(2, 1, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascan", component=yarascan.YaraScan, version=(2, 0, 0)
|
||||
),
|
||||
]
|
||||
|
||||
# get base yarascan requirements for command line options
|
||||
yarascan_requirements = yarascan.YaraScan.get_yarascan_option_requirements()
|
||||
|
||||
# return the combined requirements
|
||||
return yarascan_requirements + vadyarascan_requirements
|
||||
|
||||
@staticmethod
|
||||
def _is_syscall_block(
|
||||
disasm_func: Callable,
|
||||
syscall_finder: syscall_finder_type,
|
||||
data: bytes,
|
||||
address: int,
|
||||
) -> Optional[Tuple[str, "capstone._cs_insn"]]:
|
||||
"""
|
||||
Determines if the bytes starting at `data` represent a valid syscall instruction invocation block
|
||||
|
||||
To maliciously invoke the system call instruction, malware must do each of the following:
|
||||
|
||||
1) update RAX to the system call number
|
||||
2) update R10 to the first parameter
|
||||
3) hit the 'termination' instruction set in `syscall_finder_type`
|
||||
|
||||
We also track whether the 'syscall' instruction was encountered while parsing
|
||||
|
||||
This function is reusable for every technique we found and studied during the DEFCON research timeframe
|
||||
|
||||
Args:
|
||||
disasm_func: capstone disassembly function gathered from `get_disasm_function`
|
||||
syscall_finder: the method and constraints on the malicious system call blocks that the calling plugin knows how to find
|
||||
data: the bytes from memory to search for malicious syscall invocations
|
||||
address: the address from where `data` came from in the particular process
|
||||
Returns:
|
||||
Optional[Tuple[str, capstone._cs_insn]]: For valid blocks, the disassembled bytes in string from and the last (termination) instruction
|
||||
"""
|
||||
found_movr10 = False
|
||||
found_movreax = False
|
||||
found_syscall = False
|
||||
found_end = False
|
||||
end_inst = None
|
||||
|
||||
disasm_bytes = ""
|
||||
|
||||
for inst in disasm_func(data, address):
|
||||
disasm_bytes += f"{inst.address:#x}: {inst.mnemonic} {inst.op_str}; "
|
||||
|
||||
# an instruction of all 0x00 opcodes
|
||||
if inst.opcode.count(0) == len(inst.opcode):
|
||||
break
|
||||
|
||||
op = inst.mnemonic
|
||||
|
||||
# invalid op, bail
|
||||
if op in syscall_finder.invalid_ops:
|
||||
break
|
||||
|
||||
# found the end instruction wanted by the caller
|
||||
elif op in syscall_finder.termination_ops:
|
||||
found_end = True
|
||||
end_inst = inst
|
||||
break
|
||||
|
||||
# track this no matter what to make code more re-usable
|
||||
elif op == "syscall":
|
||||
found_syscall = True
|
||||
|
||||
# if we hit a 'syscall' but RAX or R10 haven't been touched
|
||||
# then we are in an invalid path, so bail
|
||||
if not syscall_finder.wants_syscall_inst or (
|
||||
not (found_movr10 and found_movreax)
|
||||
):
|
||||
break
|
||||
|
||||
else:
|
||||
# attempt to see if any other instruction type wrote to registers
|
||||
try:
|
||||
_, regs_written = inst.regs_access()
|
||||
except capstone.CsError:
|
||||
continue
|
||||
|
||||
if regs_written:
|
||||
for r in regs_written:
|
||||
# track writes to eax/rax or R10
|
||||
reg = inst.reg_name(r)
|
||||
if reg in ["eax", "rax"]:
|
||||
found_movreax = True
|
||||
|
||||
elif reg == "r10":
|
||||
found_movr10 = True
|
||||
|
||||
# if any of these are missing, the block is invalid regardless of
|
||||
# the technique we are trying to detect now or in the future
|
||||
if not (found_movr10 and found_movreax and found_end):
|
||||
return None
|
||||
|
||||
# if the finder requires a 'syscall' instruction then bail now if we didn't find one
|
||||
if syscall_finder.wants_syscall_inst and not found_syscall:
|
||||
return None
|
||||
|
||||
return disasm_bytes, end_inst
|
||||
|
||||
@classmethod
|
||||
def get_disasm_function(cls, architecture: str) -> Callable:
|
||||
"""
|
||||
Returns the disassembly handler for the given architecture
|
||||
.detail is used to get full instruction information
|
||||
|
||||
Args:
|
||||
architecture: the name of the architecture for the process being disassembled
|
||||
Returns:
|
||||
The disasm function from capstone for the given architecture
|
||||
"""
|
||||
disasm_types = {
|
||||
"intel": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32),
|
||||
"intel64": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64),
|
||||
}
|
||||
|
||||
disasm_type = disasm_types[architecture]
|
||||
disasm_type.detail = True
|
||||
return disasm_type.disasm
|
||||
|
||||
@classmethod
|
||||
def _is_valid_syscall(
|
||||
cls,
|
||||
syscall_finder: syscall_finder_type,
|
||||
proc_layer: interfaces.layers.DataLayerInterface,
|
||||
architecture: str,
|
||||
vads: List[Tuple[int, int, str]],
|
||||
address: int,
|
||||
) -> Optional[Tuple[int, str]]:
|
||||
"""
|
||||
Args:
|
||||
syscall_finder:
|
||||
proc_layer: the memory layer of the process being scanned
|
||||
architecture: the name of the architecture for the process being disassembled
|
||||
vads: the ranges of this process under 10MB
|
||||
address: the starting address to check for malicious syscall code blocks
|
||||
|
||||
Returns:
|
||||
Optional[Tuple[int, str]]: For valid code blocks, the starting address of the block and the disassembly string
|
||||
"""
|
||||
# the number bytes behind the yara rule hit to scan
|
||||
behind = 32
|
||||
|
||||
address = address - behind
|
||||
|
||||
try:
|
||||
data = proc_layer.read(address, behind * 2)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
disasm_func = cls.get_disasm_function(architecture)
|
||||
|
||||
# since Intel does not have fixed-size instructions, we have to scan
|
||||
# each byte offset and re-disassemble the remaining block
|
||||
for offset in range(behind):
|
||||
# if this looks like a system call back (r10, rax, ret/jmp)
|
||||
syscall_info = cls._is_syscall_block(
|
||||
disasm_func, syscall_finder, data[offset:], address + offset
|
||||
)
|
||||
if syscall_info:
|
||||
disasm_bytes, end_inst = syscall_info
|
||||
|
||||
# if we can recover (and require) a target address for this malware technique
|
||||
if syscall_finder.get_syscall_target_address:
|
||||
target_address = syscall_finder.get_syscall_target_address(
|
||||
proc_layer, end_inst
|
||||
)
|
||||
|
||||
# could not determine the address -> invalid basic block
|
||||
if not target_address:
|
||||
continue
|
||||
|
||||
# we only care about calls to system call DLLs
|
||||
path = cls.get_range_path(vads, target_address)
|
||||
if not isinstance(path, str) or not path.lower().endswith(
|
||||
cls.valid_syscall_handlers
|
||||
):
|
||||
continue
|
||||
|
||||
# return the address and disassembly string if all checks pass
|
||||
return address + offset, disasm_bytes
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_vad_maps(
|
||||
cls,
|
||||
task: interfaces.objects.ObjectInterface,
|
||||
) -> List[Tuple[int, int, str]]:
|
||||
"""Creates a map of start/end addresses within a virtual address
|
||||
descriptor tree.
|
||||
|
||||
Args:
|
||||
task: The EPROCESS object of which to traverse the vad tree
|
||||
|
||||
Returns:
|
||||
An iterable of tuples containing start and end addresses for each descriptor
|
||||
"""
|
||||
vads: List[Tuple[int, int, str]] = []
|
||||
|
||||
# scan regions under 10MB
|
||||
scan_max = 10 * 1000 * 1000
|
||||
|
||||
vad_root = task.get_vad_root()
|
||||
|
||||
for vad in vad_root.traverse():
|
||||
if vad.get_size() < scan_max:
|
||||
vads.append((vad.get_start(), vad.get_size(), vad.get_file_name()))
|
||||
|
||||
return vads
|
||||
|
||||
@classmethod
|
||||
def get_range_path(
|
||||
cls, ranges: List[Tuple[int, int, str]], address: int
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Returns the path for the range holding `address`, if found
|
||||
|
||||
Args:
|
||||
ranges: VADs collected from `get_vad_maps`
|
||||
address: the address to find
|
||||
Returns:
|
||||
The path holding the address, if any
|
||||
"""
|
||||
for start, size, path in ranges:
|
||||
if start <= address < start + size:
|
||||
return path
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_tasks_to_scan(
|
||||
cls,
|
||||
context: interfaces.context.ContextInterface,
|
||||
kernel_module_name: str,
|
||||
) -> Generator[
|
||||
Tuple[interfaces.objects.ObjectInterface, str, str, str], None, None
|
||||
]:
|
||||
"""
|
||||
Gathers active processes with the extra information needed
|
||||
to detect malicious syscall instructions
|
||||
|
||||
Returns:
|
||||
Generator of the process object, name, memory layer, and architecture
|
||||
"""
|
||||
|
||||
# gather active processes
|
||||
filter_func = pslist.PsList.create_active_process_filter()
|
||||
|
||||
kernel = context.modules[kernel_module_name]
|
||||
|
||||
is_32bit_arch = not symbols.symbol_table_is_64bit(
|
||||
context=context, symbol_table_name=kernel.symbol_table_name
|
||||
)
|
||||
|
||||
for proc in pslist.PsList.list_processes(
|
||||
context=context,
|
||||
kernel_module_name=kernel_module_name,
|
||||
filter_func=filter_func,
|
||||
):
|
||||
proc_name = utility.array_to_string(proc.ImageFileName)
|
||||
|
||||
# skip Defender
|
||||
if proc_name in ["MsMpEng.exe"]:
|
||||
continue
|
||||
|
||||
try:
|
||||
proc_layer_name = proc.add_process_layer()
|
||||
except exceptions.InvalidAddressException:
|
||||
continue
|
||||
|
||||
if is_32bit_arch or proc.get_is_wow64():
|
||||
architecture = "intel"
|
||||
else:
|
||||
architecture = "intel64"
|
||||
|
||||
yield proc, proc_name, proc_layer_name, architecture
|
||||
|
||||
@classmethod
|
||||
def _get_rule_hits(
|
||||
cls,
|
||||
context: interfaces.objects.ObjectInterface,
|
||||
proc_layer: interfaces.layers.DataLayerInterface,
|
||||
vads: List[Tuple[int, int, str]],
|
||||
pattern: str,
|
||||
) -> Generator[Tuple[int, Optional[str]], None, None]:
|
||||
"""
|
||||
Runs the given opcode rule through Yara and returns the address and file path of hits
|
||||
|
||||
Args:
|
||||
context:
|
||||
proc_layer: the layer to scan
|
||||
vads: the ranges inside of the process being scanned
|
||||
pattern: the opcodes rule from the plugin to detect a particular EDR-bypass technique
|
||||
|
||||
Returns:
|
||||
Generator of the address and file path of hits
|
||||
"""
|
||||
sections = [(vad[0], vad[1]) for vad in vads]
|
||||
|
||||
rule = yarascan.YaraScanner.get_rule(pattern)
|
||||
|
||||
for hit in proc_layer.scan(
|
||||
context=context,
|
||||
scanner=yarascan.YaraScanner(rules=rule),
|
||||
sections=sections,
|
||||
):
|
||||
address = hit[0]
|
||||
|
||||
path = cls.get_range_path(vads, address)
|
||||
|
||||
# ignore hits in the system call DLLs
|
||||
if isinstance(path, str) and path.lower().endswith(
|
||||
cls.valid_syscall_handlers
|
||||
):
|
||||
continue
|
||||
|
||||
yield address, path
|
||||
|
||||
def _generator(
|
||||
self,
|
||||
) -> Generator[Tuple[int, Tuple[str, int, Optional[str], int, str]], None, None]:
|
||||
if not has_capstone:
|
||||
vollog.warning(
|
||||
"capstone is not installed. This plugin requires capstone to operate."
|
||||
)
|
||||
return
|
||||
|
||||
for proc, proc_name, proc_layer_name, architecture in self.get_tasks_to_scan(
|
||||
self.context, self.config["kernel"]
|
||||
):
|
||||
proc_layer = self.context.layers[proc_layer_name]
|
||||
|
||||
vads = self.get_vad_maps(proc)
|
||||
if not vads:
|
||||
continue
|
||||
|
||||
# for each valid process, look for malicious syscall invocations
|
||||
for address, vad_path in self._get_rule_hits(
|
||||
self.context, proc_layer, vads, self.syscall_finder.rule_str
|
||||
):
|
||||
syscall_info = self._is_valid_syscall(
|
||||
self.syscall_finder, proc_layer, architecture, vads, address
|
||||
)
|
||||
if not syscall_info:
|
||||
continue
|
||||
|
||||
address, disasm_bytes = syscall_info
|
||||
|
||||
yield 0, (
|
||||
proc_name,
|
||||
proc.UniqueProcessId,
|
||||
vad_path,
|
||||
format_hints.Hex(address),
|
||||
disasm_bytes,
|
||||
)
|
||||
|
||||
def run(self) -> renderers.TreeGrid:
|
||||
return renderers.TreeGrid(
|
||||
[
|
||||
("Process", str),
|
||||
("PID", int),
|
||||
("Range", str),
|
||||
("Address", format_hints.Hex),
|
||||
("Disasm", str),
|
||||
],
|
||||
self._generator(),
|
||||
)
|
||||
@@ -0,0 +1,121 @@
|
||||
# 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
|
||||
#
|
||||
|
||||
import struct
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
from volatility3.framework import interfaces, exceptions
|
||||
from volatility3.framework.configuration import requirements
|
||||
from volatility3.plugins import yarascan
|
||||
from volatility3.plugins.windows.malware import direct_system_calls
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IndirectSystemCalls(direct_system_calls.DirectSystemCalls):
|
||||
"""Detects the Indirect System Call technique used to bypass EDRs."""
|
||||
|
||||
_required_framework_version = (2, 4, 0)
|
||||
_version = (1, 0, 0)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
self.syscall_finder = direct_system_calls.syscall_finder_type(
|
||||
# gets the target address of a indirect jmp
|
||||
self._indirect_syscall_block_target,
|
||||
# we are looking for indirect system calls, so we don't want 'syscall' instructions in our code block
|
||||
False,
|
||||
# jmp [address]; ret
|
||||
"/\\xff\\x25[^\\xc3]{,24}\\xc3/",
|
||||
# any of these mean we aren't in a malicious indirect call
|
||||
["call", "leave", "int3", "ret"],
|
||||
# stop at jmp, this should reference the system call instruction
|
||||
["jmp"],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
# create a list of requirements for vadyarascan
|
||||
vadyarascan_requirements = [
|
||||
requirements.ModuleRequirement(
|
||||
name="kernel",
|
||||
description="Windows kernel",
|
||||
architectures=["Intel32", "Intel64"],
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascanner", component=yarascan.YaraScanner, version=(2, 1, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascan", component=yarascan.YaraScan, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="direct_system_calls",
|
||||
component=direct_system_calls.DirectSystemCalls,
|
||||
version=(2, 0, 0),
|
||||
),
|
||||
]
|
||||
|
||||
# get base yarascan requirements for command line options
|
||||
yarascan_requirements = yarascan.YaraScan.get_yarascan_option_requirements()
|
||||
|
||||
# return the combined requirements
|
||||
return yarascan_requirements + vadyarascan_requirements
|
||||
|
||||
@staticmethod
|
||||
def _indirect_syscall_block_target(
|
||||
proc_layer: interfaces.layers.DataLayerInterface, inst
|
||||
) -> Optional[int]:
|
||||
"""
|
||||
This function determines the address of a jmp in the following form:
|
||||
|
||||
jmp [address]
|
||||
|
||||
To determine this, we must:
|
||||
1) Pull the 4 byte relative offset of 'address' inside the instruction
|
||||
2) Compute the full address of this relative offset
|
||||
3) Read from the address as it is being dereferenced
|
||||
4) Ensure the target address points to a 'syscall' instruction
|
||||
|
||||
Args:
|
||||
proc_layer: the layer of the potential syscall block
|
||||
inst: the terminating instruction of the syscall block check
|
||||
Returns:
|
||||
The target address of the jump if it can be computed
|
||||
"""
|
||||
|
||||
try:
|
||||
jmp_address_str = proc_layer.read(inst.address, 6)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# Should be an jmp...
|
||||
if jmp_address_str[0:2] != b"\xff\x25":
|
||||
return None
|
||||
|
||||
# get the address of the 'jmp [address]' instruction
|
||||
relative_offset = struct.unpack("<I", jmp_address_str[2:])[0]
|
||||
if not relative_offset or relative_offset == -1:
|
||||
return None
|
||||
|
||||
# compute the target address of the jmp (dereference)
|
||||
jmp_address = inst.address + relative_offset + 6
|
||||
try:
|
||||
jmp_target_str = proc_layer.read(jmp_address, 8)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# compute from the target address then read from it
|
||||
jmp_target_address = struct.unpack("<Q", jmp_target_str)[0]
|
||||
try:
|
||||
jmp_target = proc_layer.read(jmp_target_address, 2)
|
||||
except exceptions.InvalidAddressException:
|
||||
return None
|
||||
|
||||
# check that the address points to a 'syscall' instruction
|
||||
if jmp_target == b"\x0f\x05":
|
||||
return jmp_target_address
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user