Merge pull request #1622 from volatilityfoundation/feature/data-renderer

Feature/data renderer
This commit is contained in:
ikelos
2025-04-01 22:16:28 +01:00
committed by GitHub
12 changed files with 360 additions and 107 deletions
+110 -35
View File
@@ -9,10 +9,11 @@ import random
import string
import sys
from functools import wraps
from typing import Any, Callable, Dict, List, Tuple
from typing import Any, Callable, Dict, List, Tuple, TypeVar, Union
from volatility3.cli import text_filter
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.interfaces.renderers import BaseAbsentValue
from volatility3.framework.renderers import format_hints
vollog = logging.getLogger(__name__)
@@ -80,7 +81,10 @@ def multitypedata_as_text(value: format_hints.MultiTypeData) -> str:
return hex_bytes_as_text(value)
def optional(func: Callable) -> Callable:
T = TypeVar("T")
def optional(func: Callable[[Union[BaseAbsentValue, T]], str]) -> Callable[[T], str]:
@wraps(func)
def wrapped(x: Any) -> str:
if isinstance(x, interfaces.renderers.BaseAbsentValue):
@@ -110,7 +114,7 @@ def quoted_optional(func: Callable) -> Callable:
return wrapped
def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
def display_disassembly(disasm: renderers.Disassembly) -> str:
"""Renders a disassembly renderer type into string format.
Args:
@@ -137,9 +141,104 @@ def display_disassembly(disasm: interfaces.renderers.Disassembly) -> str:
return QuickTextRenderer._type_renderers[bytes](disasm.data)
class CLITypeRenderer(interfaces.renderers.TypeRendererInterface):
def __init__(self, func):
super().__init__(func=optional(func))
class LayerDataRenderer(CLITypeRenderer):
"""Renders a LayerData object into data/bytes"""
def __init__(self):
self.context_byte_len = 0
self.width = 16
self.display_offset = False
self.display_hex = True
self.display_ascii = True
def render(data: Union[renderers.LayerData, BaseAbsentValue]):
if isinstance(data, BaseAbsentValue):
# FIXME: Do something cleverer here
return ""
context_byte_len = self.context_byte_len if not data.no_surrounding else 0
layer = data.context.layers[data.layer_name]
# Map of the holes
error_bytes = set()
start_offset = data.offset - context_byte_len
end_offset = data.offset + data.length + context_byte_len
if isinstance(layer, interfaces.layers.TranslationLayerInterface):
error_bytes = set()
mapping = iter(layer.mapping(start_offset, end_offset, True))
current_map = next(mapping)
for i in range(start_offset, end_offset):
# Run through the bytes, check if they're present
offset, sublength, _, _, _ = current_map
if i < offset:
error_bytes.add(i - start_offset)
if i > offset + sublength:
try:
current_map = next(mapping)
except StopIteration:
pass
offset, sublength, _, _, _ = current_map
if i > offset + sublength:
error_bytes.add(i - start_offset)
# Padded data
specific_data = data.context.layers[data.layer_name].read(
start_offset,
end_offset - start_offset,
True,
)
printables = ""
output = "\n"
for count, byte in enumerate(specific_data):
if count not in error_bytes:
output += f"{byte:02x} "
char = chr(byte)
printables += char if 0x20 <= byte <= 0x7E else "."
else:
output += "__ "
printables += "."
if count % self.width == self.width - 1:
output += printables
if count < len(specific_data) - 1:
output += "\n"
printables = ""
# Handle leftovers when the length is not mutiple of width
if printables:
padding = self.width - len(printables)
output += " " * padding
output += printables
output += " " * padding
return output
render_func = render
return super().__init__(render_func)
class CLIRenderer(interfaces.renderers.Renderer):
"""Class to add specific requirements for CLI renderers."""
_type_renderers = {
format_hints.Bin: CLITypeRenderer(lambda x: f"0b{x:b}"),
format_hints.Hex: CLITypeRenderer(lambda x: f"0x{x:x}"),
format_hints.HexBytes: CLITypeRenderer(hex_bytes_as_text),
format_hints.MultiTypeData: CLITypeRenderer(multitypedata_as_text),
renderers.Disassembly: CLITypeRenderer(display_disassembly),
bytes: CLITypeRenderer(lambda x: " ".join(f"{b:02x}" for b in x)),
renderers.LayerData: LayerDataRenderer(),
datetime.datetime: CLITypeRenderer(
lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")
),
"default": CLITypeRenderer(lambda x: f"{x}"),
}
name = "unnamed"
structured_output = False
filter: text_filter.CLIFilter = None
@@ -170,21 +269,11 @@ class CLIRenderer(interfaces.renderers.Renderer):
class QuickTextRenderer(CLIRenderer):
_type_renderers = {
format_hints.Bin: optional(lambda x: f"0b{x:b}"),
format_hints.Hex: optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: optional(hex_bytes_as_text),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
interfaces.renderers.Disassembly: optional(display_disassembly),
bytes: optional(lambda x: " ".join(f"{b:02x}" for b in x)),
datetime.datetime: optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
"default": optional(lambda x: f"{x}"),
}
name = "quick"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each column immediately to stdout.
@@ -242,7 +331,7 @@ class NoneRenderer(CLIRenderer):
name = "none"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
if not grid.populated:
@@ -250,22 +339,12 @@ class NoneRenderer(CLIRenderer):
class CSVRenderer(CLIRenderer):
_type_renderers = {
format_hints.Bin: optional(lambda x: f"0b{x:b}"),
format_hints.Hex: optional(lambda x: f"0x{x:x}"),
format_hints.HexBytes: optional(hex_bytes_as_text),
format_hints.MultiTypeData: optional(multitypedata_as_text),
interfaces.renderers.Disassembly: optional(display_disassembly),
bytes: optional(lambda x: " ".join(f"{b:02x}" for b in x)),
datetime.datetime: optional(lambda x: x.strftime("%Y-%m-%d %H:%M:%S.%f %Z")),
"default": optional(lambda x: f"{x}"),
}
name = "csv"
structured_output = True
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each row immediately to stdout.
@@ -316,12 +395,10 @@ class CSVRenderer(CLIRenderer):
class PrettyTextRenderer(CLIRenderer):
_type_renderers = QuickTextRenderer._type_renderers
name = "pretty"
def get_render_options(self):
pass
return []
def render(self, grid: interfaces.renderers.TreeGrid) -> None:
"""Renders each column immediately to stdout.
@@ -380,7 +457,7 @@ class PrettyTextRenderer(CLIRenderer):
accumulator.append((node.path_depth, line))
return accumulator
final_output: List[Tuple[int, Dict[interfaces.renderers.Column, bytes]]] = []
final_output: List[Tuple[int, Dict[interfaces.renderers.Column, str]]] = []
if not grid.populated:
grid.populate(visitor, final_output)
else:
@@ -417,9 +494,7 @@ class PrettyTextRenderer(CLIRenderer):
if column in ignore_columns:
del line[column]
else:
line[column] = line[column] + (
[""] * (nums_line - len(line[column]))
)
line[column] = line[column] + ("" * (nums_line - len(line[column])))
for index in range(nums_line):
if index == 0:
outfd.write(
@@ -448,7 +523,7 @@ class PrettyTextRenderer(CLIRenderer):
class JsonRenderer(CLIRenderer):
_type_renderers = {
format_hints.HexBytes: quoted_optional(hex_bytes_as_text),
interfaces.renderers.Disassembly: quoted_optional(display_disassembly),
renderers.Disassembly: quoted_optional(display_disassembly),
format_hints.MultiTypeData: quoted_optional(multitypedata_as_text),
bytes: optional(lambda x: " ".join(f"{b:02x}" for b in x)),
datetime.datetime: lambda x: (
@@ -463,7 +538,7 @@ class JsonRenderer(CLIRenderer):
structured_output = True
def get_render_options(self) -> List[interfaces.renderers.RenderOption]:
pass
return []
def output_result(self, outfd, result):
"""Outputs the JSON data to a file in a particular format"""
+1 -1
View File
@@ -1,6 +1,6 @@
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 25 # Number of changes that only add to the interface
VERSION_MINOR = 26 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
+74 -18
View File
@@ -10,7 +10,8 @@ suitable output.
"""
import datetime
from abc import abstractmethod, ABCMeta
import warnings
from abc import ABCMeta, abstractmethod
from collections import abc
from typing import (
Any,
@@ -20,11 +21,28 @@ from typing import (
List,
NamedTuple,
Optional,
TypeVar,
Type,
Tuple,
Type,
TypeVar,
Union,
)
from typing import Dict
from volatility3.framework import interfaces
class BasicType:
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return str(self)
class BaseAbsentValue:
"""Class that represents values which are not present for some reason."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class Column(NamedTuple):
@@ -34,11 +52,37 @@ class Column(NamedTuple):
RenderOption = Any
T = TypeVar("T")
class TypeRendererInterface:
type = T
def __init__(
self, func: Optional[Callable] = None, options: Optional[Dict[str, Any]] = None
):
self._options = options or {}
setattr(self, "render", func)
@property
def options(self):
return self._options
def render(self, data: Union[T, BaseAbsentValue]) -> Any:
"""Renders a specific datatype"""
return ""
def __call__(self, data: Union[T, BaseAbsentValue]) -> Any:
"""Shortcut for render"""
return self.render(data)
class Renderer(metaclass=ABCMeta):
"""Class that defines the interface that all output renderers must
support."""
_type_renderers: Dict[Union[Type, str], Callable]
def __init__(self, options: Optional[List[RenderOption]] = None) -> None:
"""Accepts an options object to configure the renderers."""
# FIXME: Once the config option objects are in place, put the _type_check in place
@@ -102,11 +146,7 @@ class TreeNode(abc.Sequence, metaclass=ABCMeta):
"""
class BaseAbsentValue:
"""Class that represents values which are not present for some reason."""
class Disassembly:
class Disassembly(BasicType):
"""A class to indicate that the bytes provided should be disassembled
(based on the architecture)"""
@@ -115,6 +155,10 @@ class Disassembly:
def __init__(
self, data: bytes, offset: int = 0, architecture: str = "intel64"
) -> None:
warnings.warn(
"interfaces.renderers.Disassembly is now renderers.Disassembly",
FutureWarning,
)
self.data = data
self.architecture = None
if architecture in self.possible_architectures:
@@ -123,6 +167,10 @@ class Disassembly:
raise TypeError("Offset must be an integer type")
self.offset = offset
def __str__(self) -> str:
"""Fallback method of rendering"""
return str(self.data)
# We don't class these off a shared base, because the BaseTypes must only
# contain the types that the validator will accept (which would not include the base)
@@ -135,7 +183,7 @@ BaseTypes = Union[
Type[bytes],
Type[datetime.datetime],
Type[BaseAbsentValue],
Type[Disassembly],
Type[BasicType],
]
ColumnsType = List[Tuple[str, BaseTypes]]
VisitorSignature = Callable[[TreeNode, _Type], _Type]
@@ -154,16 +202,15 @@ class TreeGrid(metaclass=ABCMeta):
and to create cycles.
"""
base_types: ClassVar[Tuple] = (
int,
str,
float,
bytes,
datetime.datetime,
Disassembly,
)
# TODO: Figure out why this isn't just BaseTypes (which includes AbsentValues'
base_types: ClassVar[Tuple] = (int, str, float, bytes, datetime.datetime, BasicType)
def __init__(self, columns: ColumnsType, generator: Generator) -> None:
def __init__(
self,
columns: ColumnsType,
generator: Generator,
context: Optional["interfaces.context.ContextInterface"] = None,
) -> None:
"""Constructs a TreeGrid object using a specific set of columns.
The TreeGrid itself is a root element, that can have children but no values.
@@ -174,6 +221,15 @@ class TreeGrid(metaclass=ABCMeta):
columns: A list of column tuples made up of (name, type).
generator: An iterable containing row for a tree grid, each row contains a indent level followed by the values for each column in order.
"""
self._context = context
@property
def context(self) -> Optional["interfaces.context.ContextInterface"]:
"""Returns the context value for the tree grid (to retrieve data items)
This is a property to ensure the renderers don't try changing the context for any reason
"""
return self._context
@staticmethod
@abstractmethod
@@ -18,7 +18,7 @@ class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 2)
_version = (1, 0, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -76,9 +76,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vma.vm_start, architecture
)
disasm = renderers.Disassembly(data, vma.vm_start, architecture)
yield (
0,
@@ -106,7 +104,7 @@ class Malfind(interfaces.plugins.PluginInterface):
("Path", str),
("Protection", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(
pslist.PsList.list_tasks(
@@ -17,8 +17,8 @@ vollog = logging.getLogger(__name__)
class VmaYaraScan(interfaces.plugins.PluginInterface):
"""Scans all virtual memory areas for tasks using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 0, 3)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 4)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -97,12 +97,18 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
for offset, rule_name, name, value in scanner(
proc_layer.read(start, size, pad=True), start
):
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=proc_layer.name,
length=len(value),
)
yield 0, (
format_hints.Hex(offset),
task.tgid,
rule_name,
name,
value,
layer_data,
)
@classmethod
@@ -130,7 +136,7 @@ class VmaYaraScan(interfaces.plugins.PluginInterface):
("PID", int),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
+3 -5
View File
@@ -13,7 +13,7 @@ from volatility3.plugins.mac import pslist
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 0, 0)
_required_framework_version = (2, 0, 1)
@classmethod
def get_requirements(cls):
@@ -68,9 +68,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vma.links.start, architecture
)
disasm = renderers.Disassembly(data, vma.links.start, architecture)
yield (
0,
@@ -99,7 +97,7 @@ class Malfind(interfaces.plugins.PluginInterface):
("End", format_hints.Hex),
("Protection", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(
list_tasks(self.context, self.config["kernel"], filter_func=filter_func)
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import Iterable, Tuple
from typing import Iterable, Generator, Tuple
from volatility3.framework import interfaces, symbols, exceptions
from volatility3.framework import renderers
@@ -17,7 +17,8 @@ vollog = logging.getLogger(__name__)
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code."""
_required_framework_version = (2, 4, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 1, 0)
@classmethod
def get_requirements(cls):
@@ -88,6 +89,25 @@ class Malfind(interfaces.plugins.PluginInterface):
symbol_table: str,
proc: interfaces.objects.ObjectInterface,
) -> Iterable[Tuple[interfaces.objects.ObjectInterface, bytes]]:
for vad, data_object in cls.list_injection_sites(
context, kernel_layer_name, symbol_table, proc
):
yield vad, data_object.context.layers[data_object.layer_name].read(
data_object.offset, data_object.length
)
@classmethod
def list_injection_sites(
cls,
context: interfaces.context.ContextInterface,
kernel_layer_name: str,
symbol_table: str,
proc: interfaces.objects.ObjectInterface,
) -> Generator[
Tuple[interfaces.objects.ObjectInterface, renderers.LayerData],
None,
None,
]:
"""Generate memory regions for a process that may contain injected
code.
@@ -156,8 +176,16 @@ class Malfind(interfaces.plugins.PluginInterface):
vollog.warning(
f"[proc_id {proc_id}] Found suspicious DIRTY + {protection_string} page at {hex(dirty_page)}",
)
data = proc_layer.read(vad.get_start(), 64, pad=True)
yield vad, data
start = vad.get_start()
length = 64
data = renderers.LayerData(
context=context,
layer_name=proc_layer_name,
offset=start,
length=length,
no_surrounding=True,
)
yield (vad, data)
def _generator(self, procs):
# determine if we're on a 32 or 64 bit kernel
@@ -166,7 +194,7 @@ class Malfind(interfaces.plugins.PluginInterface):
# set refined criteria to know when to add to "Notes" column
refined_criteria = {
b"MZ": "MZ header",
b"\x55\x8B": "PE header",
b"\x55\x8b": "PE header",
b"\x55\x48": "Function prologue",
b"\x55\x89": "Function prologue",
}
@@ -179,11 +207,14 @@ class Malfind(interfaces.plugins.PluginInterface):
# by default, "Notes" column will be set to N/A
process_name = utility.array_to_string(proc.ImageFileName)
for vad, data in self.list_injections(
for vad, data_object in self.list_injection_sites(
self.context, kernel.layer_name, kernel.symbol_table_name, proc
):
notes = renderers.NotApplicableValue()
# Check for unique headers and update "Notes" column if criteria is met
data = data_object.context.layers[data_object.layer_name].read(
data_object.offset, data_object.length, True
)
if data[0:2] in refined_criteria:
notes = refined_criteria[data[0:2]]
@@ -193,9 +224,7 @@ class Malfind(interfaces.plugins.PluginInterface):
else:
architecture = "intel64"
disasm = interfaces.renderers.Disassembly(
data, vad.get_start(), architecture
)
disasm = renderers.Disassembly(data, vad.get_start(), architecture)
file_output = "Disabled"
if self.config["dump"]:
@@ -231,7 +260,7 @@ class Malfind(interfaces.plugins.PluginInterface):
vad.get_private_memory(),
file_output,
notes,
format_hints.HexBytes(data),
data_object,
disasm,
),
)
@@ -251,8 +280,8 @@ class Malfind(interfaces.plugins.PluginInterface):
("PrivateMemory", int),
("File output", str),
("Notes", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces.renderers.Disassembly),
("Hexdump", renderers.LayerData),
("Disasm", renderers.Disassembly),
],
self._generator(
pslist.PsList.list_processes(
@@ -20,8 +20,8 @@ vollog = logging.getLogger(__name__)
class MBRScan(interfaces.plugins.PluginInterface):
"""Scans for and parses potential Master Boot Records (MBRs)"""
_required_framework_version = (2, 0, 1)
_version = (1, 0, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -79,7 +79,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
partition_table_object = symbol_table + constants.BANG + "PARTITION_TABLE"
# Define Signature and Data Length
mbr_signature = b"\x55\xAA"
mbr_signature = b"\x55\xaa"
mbr_length = 0x200
bootcode_length = 0x1B8
@@ -125,9 +125,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(
bootcode, 0, architecture
),
renderers.Disassembly(bootcode, 0, architecture),
),
)
else:
@@ -151,10 +149,14 @@ class MBRScan(interfaces.plugins.PluginInterface):
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
renderers.NotApplicableValue(),
interfaces.renderers.Disassembly(
bootcode, 0, architecture
renderers.Disassembly(bootcode, 0, architecture),
renderers.LayerData(
context=self.context,
layer_name=layer.name,
offset=mbr_start_offset,
length=bootcode_length,
no_surrounding=True,
),
format_hints.HexBytes(bootcode),
),
)
@@ -237,7 +239,7 @@ class MBRScan(interfaces.plugins.PluginInterface):
("Bootable", bool),
("PartitionType", str),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly),
("Disasm", renderers.Disassembly),
],
self._generator(),
)
@@ -261,8 +263,8 @@ class MBRScan(interfaces.plugins.PluginInterface):
("EndingCHS", int),
("EndingSector", int),
("SectorInSize", format_hints.Hex),
("Disasm", interfaces.renderers.Disassembly),
("Bootcode", format_hints.HexBytes),
("Disasm", renderers.Disassembly),
("Bootcode", renderers.LayerData),
],
self._generator(),
)
@@ -231,7 +231,7 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
content = attr.get_resident_filecontent()
if content:
content = format_hints.HexBytes(content)
content = renderers.LayerData.from_object(content)
else:
content = renderers.NotAvailableValue()
@@ -334,9 +334,9 @@ class MFTScan(interfaces.plugins.PluginInterface, timeliner.TimeLinerInterface):
class ADS(interfaces.plugins.PluginInterface):
"""Scans for Alternate Data Stream"""
_required_framework_version = (2, 7, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
@@ -395,7 +395,7 @@ class ADS(interfaces.plugins.PluginInterface):
("MFT Type", str),
("Filename", str),
("ADS Filename", str),
("Hexdump", format_hints.HexBytes),
("Hexdump", renderers.LayerData),
],
self._generator(),
)
@@ -404,9 +404,9 @@ class ADS(interfaces.plugins.PluginInterface):
class ResidentData(interfaces.plugins.PluginInterface):
"""Scans for MFT Records with Resident Data"""
_required_framework_version = (2, 7, 0)
_required_framework_version = (2, 22, 0)
_version = (1, 0, 1)
_version = (1, 0, 2)
@classmethod
def get_requirements(cls):
@@ -461,7 +461,7 @@ class ResidentData(interfaces.plugins.PluginInterface):
("Record Number", int),
("MFT Type", str),
("Filename", str),
("Hexdump", format_hints.HexBytes),
("Hexdump", renderers.LayerData),
],
self._generator(),
)
@@ -17,8 +17,8 @@ vollog = logging.getLogger(__name__)
class VadYaraScan(interfaces.plugins.PluginInterface):
"""Scans all the Virtual Address Descriptor memory maps using yara."""
_required_framework_version = (2, 4, 0)
_version = (1, 1, 2)
_required_framework_version = (2, 22, 0)
_version = (1, 1, 3)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -93,12 +93,18 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
for offset, rule_name, name, value in scanner(
layer.read(start, size, pad=True), start
):
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=layer.name,
length=len(value),
)
yield 0, (
format_hints.Hex(offset),
task.UniqueProcessId,
rule_name,
name,
value,
layer_data,
)
@classmethod
@@ -126,7 +132,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
("PID", int),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
+10 -4
View File
@@ -105,8 +105,8 @@ class YaraScanner(interfaces.layers.ScannerInterface):
class YaraScan(plugins.PluginInterface):
"""Scans kernel memory using yara rules (string or file)."""
_required_framework_version = (2, 0, 0)
_version = (2, 0, 0)
_required_framework_version = (2, 22, 0)
_version = (2, 0, 1)
_yara_x = USE_YARA_X
@classmethod
@@ -206,7 +206,13 @@ class YaraScan(plugins.PluginInterface):
for offset, rule_name, name, value in layer.scan(
context=self.context, scanner=YaraScanner(rules=rules)
):
yield 0, (format_hints.Hex(offset), rule_name, name, value)
layer_data = renderers.LayerData(
context=self.context,
offset=offset,
layer_name=layer.name,
length=len(value),
)
yield 0, (format_hints.Hex(offset), rule_name, name, layer_data)
def run(self):
return renderers.TreeGrid(
@@ -214,7 +220,7 @@ class YaraScan(plugins.PluginInterface):
("Offset", format_hints.Hex),
("Rule", str),
("Component", str),
("Value", bytes),
("Value", renderers.LayerData),
],
self._generator(),
)
@@ -8,6 +8,7 @@ or file or graphical output
"""
import collections
import collections.abc
import dataclasses
import datetime
import logging
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar, Union
@@ -22,16 +23,28 @@ class UnreadableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because the data cannot be
read."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class UnparsableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because the data cannot be
interpreted correctly."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "-"
class NotApplicableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which are empty because they don't make
sense for this node."""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "N/A"
class NotAvailableValue(interfaces.renderers.BaseAbsentValue):
"""Class that represents values which cannot be provided now (but might in
@@ -45,6 +58,70 @@ class NotAvailableValue(interfaces.renderers.BaseAbsentValue):
in preference, and only if neither fits should this be used.
"""
def __str__(self) -> str:
"""Fallback method for rendering basic types"""
return "N/A"
##########
### Basic Types
class Disassembly(interfaces.renderers.BasicType):
"""A class to indicate that the bytes provided should be disassembled
(based on the architecture)"""
possible_architectures = ["intel", "intel64", "arm", "arm64"]
def __init__(
self, data: bytes, offset: int = 0, architecture: str = "intel64"
) -> None:
self.data = data
self.architecture = None
if architecture in self.possible_architectures:
self.architecture = architecture
if not isinstance(offset, int):
raise TypeError("Offset must be an integer type")
self.offset = offset
def __str__(self) -> str:
"""Fallback method of rendering"""
return str(self.data)
@dataclasses.dataclass
class LayerData(interfaces.renderers.BasicType):
"""Layer data
This requires the contex to be passed in, in case plugins want to use multiple contexts
and to ensure the TreeGrid interface doesn't change, since this would break all existing plugins
"""
context: "interfaces.context.ContextInterface"
layer_name: str
offset: int
length: int
no_surrounding: bool = False
@staticmethod
def from_object(
object: "interfaces.objects.ObjectInterface",
size: Optional[int] = None,
no_surrounding: bool = True,
):
return LayerData(
context=object._context,
layer_name=object.vol.layer_name,
offset=object.vol.offset,
length=size or object.vol.size,
no_surrounding=no_surrounding,
)
def __str__(self) -> str:
"""Fallback method of rendering"""
data = self.context.layers[self.layer_name].read(self.offset, self.length, True)
return str(data)
class TreeNode(interfaces.renderers.TreeNode):
"""Class representing a particular node in a tree grid."""