mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-07 10:17:38 +02:00
Fix up vmayarascan and vadyarascan to use yarascan properly
This commit is contained in:
Mike Auty
@@ -2,6 +2,7 @@
|
||||
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
|
||||
#
|
||||
|
||||
import logging
|
||||
from typing import Iterable, List, Tuple
|
||||
|
||||
from volatility3.framework import interfaces, renderers
|
||||
@@ -10,6 +11,8 @@ from volatility3.framework.renderers import format_hints
|
||||
from volatility3.plugins import yarascan
|
||||
from volatility3.plugins.linux import pslist
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class VmaYaraScan(interfaces.plugins.PluginInterface):
|
||||
"""Scans all virtual memory areas for tasks using yara."""
|
||||
@@ -50,6 +53,8 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
|
||||
# use yarascan to parse the yara options provided and create the rules
|
||||
rules = yarascan.YaraScan.process_yara_options(dict(self.config))
|
||||
|
||||
sanity_check = 1024 * 1024 * 1024 # 1 GB
|
||||
|
||||
# filter based on the pid option if provided
|
||||
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
|
||||
for task in pslist.PsList.list_tasks(
|
||||
@@ -66,29 +71,36 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
|
||||
# get the proc_layer object from the context
|
||||
proc_layer = self.context.layers[proc_layer_name]
|
||||
|
||||
for start, end in self.get_vma_maps(task):
|
||||
for match in rules.match(
|
||||
data=proc_layer.read(start, end - start, True)
|
||||
):
|
||||
if yarascan.YaraScan.yara_returns_instances():
|
||||
for match_string in match.strings:
|
||||
for instance in match_string.instances:
|
||||
yield 0, (
|
||||
format_hints.Hex(instance.offset + start),
|
||||
task.UniqueProcessId,
|
||||
match.rule,
|
||||
match_string.identifier,
|
||||
instance.matched_data,
|
||||
)
|
||||
else:
|
||||
for offset, name, value in match.strings:
|
||||
yield 0, (
|
||||
format_hints.Hex(offset + start),
|
||||
task.tgid,
|
||||
match.rule,
|
||||
name,
|
||||
value,
|
||||
)
|
||||
vma_maps = list(self.get_vma_maps(task))
|
||||
insane_vma_maps = [
|
||||
start for (start, size) in vma_maps if size > sanity_check
|
||||
]
|
||||
for start in insane_vma_maps:
|
||||
vollog.debug(f"VMA at 0x{start:x} over sanity-check size, not scanning")
|
||||
|
||||
if not vma_maps:
|
||||
vollog.warning(f"No VMAs were found for task {task.pid}, aborting")
|
||||
continue
|
||||
|
||||
max_vma_size: int = max(
|
||||
[size for (start, size) in vma_maps if size <= sanity_check]
|
||||
)
|
||||
scanner = yarascan.YaraScanner(rules=rules)
|
||||
scanner.chunk_size = max_vma_size
|
||||
|
||||
# scan the process layer with the yarascanner
|
||||
for offset, rule_name, name, value in proc_layer.scan(
|
||||
context=self.context,
|
||||
scanner=scanner,
|
||||
sections=vma_maps,
|
||||
):
|
||||
yield 0, (
|
||||
format_hints.Hex(offset),
|
||||
task.tgid,
|
||||
rule_name,
|
||||
name,
|
||||
value,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_vma_maps(
|
||||
|
||||
@@ -32,6 +32,9 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
|
||||
requirements.PluginRequirement(
|
||||
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
|
||||
),
|
||||
requirements.VersionRequirement(
|
||||
name="yarascanner", component=yarascan.YaraScanner, version=(2, 0, 0)
|
||||
),
|
||||
requirements.PluginRequirement(
|
||||
name="yarascan", plugin=yarascan.YaraScan, version=(2, 0, 0)
|
||||
),
|
||||
@@ -66,49 +69,33 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
|
||||
):
|
||||
layer_name = task.add_process_layer()
|
||||
layer = self.context.layers[layer_name]
|
||||
for start, size in self.get_vad_maps(task):
|
||||
if size > sanity_check:
|
||||
vollog.debug(
|
||||
f"VAD at 0x{start:x} over sanity-check size, not scanning"
|
||||
)
|
||||
continue
|
||||
|
||||
data = layer.read(start, size, True)
|
||||
if not yarascan.YaraScan._yara_x:
|
||||
for match in rules.match(data=data):
|
||||
if yarascan.YaraScan.yara_returns_instances():
|
||||
for match_string in match.strings:
|
||||
for instance in match_string.instances:
|
||||
yield 0, (
|
||||
format_hints.Hex(instance.offset + start),
|
||||
task.UniqueProcessId,
|
||||
match.rule,
|
||||
match_string.identifier,
|
||||
instance.matched_data,
|
||||
)
|
||||
else:
|
||||
for offset, name, value in match.strings:
|
||||
yield 0, (
|
||||
format_hints.Hex(offset + start),
|
||||
task.UniqueProcessId,
|
||||
match.rule,
|
||||
name,
|
||||
value,
|
||||
)
|
||||
else:
|
||||
for match in rules.scan(data).matching_rules:
|
||||
for match_string in match.patterns:
|
||||
for instance in match_string.matches:
|
||||
yield 0, (
|
||||
format_hints.Hex(instance.offset + start),
|
||||
task.UniqueProcessId,
|
||||
f"{match.namespace}.{match.identifier}",
|
||||
match_string.identifier,
|
||||
data[
|
||||
instance.offset : instance.offset
|
||||
+ instance.length
|
||||
],
|
||||
)
|
||||
vad_maps = list(self.get_vad_maps(task))
|
||||
insane_vad_maps = [
|
||||
start for (start, size) in vad_maps if size > sanity_check
|
||||
]
|
||||
for start in insane_vad_maps:
|
||||
vollog.debug(f"VAD at 0x{start:x} over sanity-check size, not scanning")
|
||||
|
||||
max_vad_size: int = max(
|
||||
[size for (start, size) in vad_maps if size <= sanity_check]
|
||||
)
|
||||
scanner = yarascan.YaraScanner(rules=rules)
|
||||
scanner.chunk_size = max_vad_size
|
||||
|
||||
# scan the process layer with the yarascanner
|
||||
for offset, rule_name, name, value in layer.scan(
|
||||
context=self.context,
|
||||
scanner=scanner,
|
||||
sections=vad_maps,
|
||||
):
|
||||
yield 0, (
|
||||
format_hints.Hex(offset),
|
||||
task.UniqueProcessId,
|
||||
rule_name,
|
||||
name,
|
||||
value,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_vad_maps(
|
||||
|
||||
Reference in New Issue
Block a user