mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-22 17:44:52 +02:00
Rework how the plugin_version is checked.
This commit is contained in:
@@ -25,7 +25,7 @@ etc) as well as indicating what they expect to be in the context (such as partic
|
||||
"""
|
||||
import abc
|
||||
import logging
|
||||
from typing import Any, ClassVar, List, Optional, Type, Dict
|
||||
from typing import Any, ClassVar, List, Optional, Type, Dict, Tuple
|
||||
|
||||
from volatility.framework import constants, interfaces
|
||||
from volatility.framework.interfaces import configuration
|
||||
@@ -366,3 +366,29 @@ class SymbolTableRequirement(configuration.ConstructableRequirementInterface,
|
||||
value: Any) -> configuration.HierarchicalDict:
|
||||
"""Builds the appropriate configuration for the specified requirement"""
|
||||
return context.symbol_space[value].build_configuration()
|
||||
|
||||
|
||||
class PluginRequirement(interfaces.configuration.RequirementInterface):
|
||||
|
||||
def __init__(self,
|
||||
name: str,
|
||||
description: str = None,
|
||||
default: None = None,
|
||||
optional: bool = False,
|
||||
plugin: interfaces.plugins.PluginInterface = None,
|
||||
version: Optional[Tuple[int, ...]] = None) -> None:
|
||||
super().__init__(name = name, description = description, default = default, optional = optional)
|
||||
if plugin is None:
|
||||
raise ValueError("Plugin cannot be None")
|
||||
self._plugin = plugin
|
||||
if version is None:
|
||||
raise ValueError("Version cannot be None")
|
||||
self._version = version
|
||||
|
||||
def unsatisfied(self, context: interfaces.context.ContextInterface,
|
||||
config_path: str) -> Dict[str, interfaces.configuration.RequirementInterface]:
|
||||
if len(self._version) > 0 and self._plugin.version[0] != self._version[0]:
|
||||
return {config_path: self}
|
||||
if len(self._version) > 1 and self._plugin.version[1] > self._version[1]:
|
||||
return {config_path: self}
|
||||
return {}
|
||||
|
||||
@@ -113,31 +113,6 @@ class PluginInterface(interfaces_configuration.ConfigurableInterface, metaclass
|
||||
"""
|
||||
return (0, 0, 0)
|
||||
|
||||
@classmethod
|
||||
def check_plugin_version(cls, plugin: 'PluginInterface', required_version: Tuple[int, ...]) -> bool:
|
||||
"""Verify the plugin provides the needed API.
|
||||
Validating the existence of the plugin is handled by the import statement and python's traditional import machinery
|
||||
|
||||
Args
|
||||
plugin: The plugin whose version needs checking
|
||||
required_version: Tuple of the minimum semantic version required for the plugin
|
||||
"""
|
||||
result = True
|
||||
plugin_name = plugin.__name__
|
||||
if len(required_version) > 0 and plugin.version[0] != required_version[0]:
|
||||
raise exceptions.PluginVersionException("Version {} of {} does not meet required version {}".format(
|
||||
plugin.version[0], plugin_name, required_version[0]))
|
||||
if len(required_version) > 1 and plugin.version[1] > required_version[1]:
|
||||
raise exceptions.PluginVersionException("Version {}.{} of {} does not meet required version {}.{}".format(
|
||||
plugin.version[0], plugin.version[1], plugin_name, required_version[0], required_version[1]))
|
||||
if len(required_version) > 2 and plugin.version[1] == required_version[1] and plugin.version[2] > \
|
||||
required_version[2]:
|
||||
raise exceptions.PluginVersionException(
|
||||
"Version {}.{}.{} of {} does not meet required version {}.{}.{}".format(
|
||||
plugin.version[0], plugin.version[1], plugin.version[2], plugin_name, required_version[0],
|
||||
required_version[1], required_version[2]))
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces_configuration.RequirementInterface]:
|
||||
"""Returns a list of Requirement objects for this plugin"""
|
||||
|
||||
@@ -37,6 +37,8 @@ class CmdLine(interfaces_plugins.PluginInterface):
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
requirements.PluginRequirement(
|
||||
name = 'pslist', description = 'PsList plugin requirement', plugin = pslist.PsList, version = (1, 0, 0))
|
||||
requirements.IntRequirement(
|
||||
name = 'pid', description = "Process ID to include (all other processes are excluded)", optional = True)
|
||||
]
|
||||
@@ -65,8 +67,6 @@ class CmdLine(interfaces_plugins.PluginInterface):
|
||||
yield (0, (proc.UniqueProcessId, process_name, result_text))
|
||||
|
||||
def run(self):
|
||||
self.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
|
||||
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Args", str)],
|
||||
|
||||
@@ -37,6 +37,7 @@ class DllList(interfaces_plugins.PluginInterface):
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0))
|
||||
requirements.IntRequirement(
|
||||
name = 'pid', description = "Process ID to include (all other processes are excluded)", optional = True)
|
||||
]
|
||||
@@ -62,9 +63,6 @@ class DllList(interfaces_plugins.PluginInterface):
|
||||
FullDllName))
|
||||
|
||||
def run(self):
|
||||
|
||||
self.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
|
||||
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
|
||||
|
||||
return renderers.TreeGrid([("PID", int), ("Process", str), ("Base", format_hints.Hex),
|
||||
|
||||
@@ -42,6 +42,8 @@ class DriverIrp(plugins.PluginInterface):
|
||||
@classmethod
|
||||
def get_requirements(cls):
|
||||
return [
|
||||
requirements.PluginRequirement(name = 'ssdt', plugin = ssdt.SSDT, version = (1, 0, 0)),
|
||||
requirements.PluginRequirement(name = 'driverscan', plugin = driverscan.DriverScan, version = (1, 0, 0)),
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
@@ -75,8 +77,6 @@ class DriverIrp(plugins.PluginInterface):
|
||||
format_hints.Hex(address), module_name, renderers.NotAvailableValue()))
|
||||
|
||||
def run(self):
|
||||
self.check_plugin_version(ssdt.SSDT, (1, 0, 0))
|
||||
self.check_plugin_version(driverscan.DriverScan, (1, 0, 0))
|
||||
|
||||
return renderers.TreeGrid([
|
||||
("Offset", format_hints.Hex),
|
||||
|
||||
@@ -40,6 +40,7 @@ class ModDump(interfaces.plugins.PluginInterface):
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
# Reuse the requirements from the plugins we use
|
||||
return [
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols")
|
||||
@@ -58,7 +59,6 @@ class ModDump(interfaces.plugins.PluginInterface):
|
||||
Returns:
|
||||
<list> of layer names
|
||||
"""
|
||||
cls.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
|
||||
filter_func = pslist.PsList.create_pid_filter(pids or [])
|
||||
|
||||
|
||||
@@ -25,8 +25,8 @@ import volatility.framework.interfaces.plugins as plugins
|
||||
from volatility import classproperty
|
||||
from volatility.framework import renderers, interfaces, layers
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.framework.objects import utility
|
||||
from volatility.framework.renderers import format_hints
|
||||
from volatility.plugins import timeliner
|
||||
|
||||
|
||||
@@ -53,7 +53,7 @@ class PsList(plugins.PluginInterface, timeliner.TimeLinerInterface):
|
||||
|
||||
@classproperty
|
||||
def version(cls):
|
||||
return (2, 0, 0)
|
||||
return (1, 0, 0)
|
||||
|
||||
@classmethod
|
||||
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[interfaces.objects.ObjectInterface], bool]:
|
||||
|
||||
@@ -36,6 +36,7 @@ class Strings(interfaces.plugins.PluginInterface):
|
||||
@classmethod
|
||||
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
|
||||
return [
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
@@ -44,8 +45,6 @@ class Strings(interfaces.plugins.PluginInterface):
|
||||
# TODO: Make URLRequirement that can accept a file address which the framework can open
|
||||
|
||||
def run(self):
|
||||
self.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
|
||||
return renderers.TreeGrid([("String", str), ("Physical Address", format_hints.Hex), ("Result", str)],
|
||||
self._generator())
|
||||
|
||||
|
||||
@@ -54,6 +54,7 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
|
||||
default = 0x40000000,
|
||||
description = "Set the maximum size (default is 1GB)",
|
||||
optional = True),
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0))
|
||||
requirements.IntRequirement(
|
||||
name = 'pid', description = "Process ID to include (all other processes are excluded)", optional = True)
|
||||
]
|
||||
@@ -99,6 +100,4 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
|
||||
yield (start, end - start)
|
||||
|
||||
def run(self):
|
||||
self.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
|
||||
return renderers.TreeGrid([('Offset', format_hints.Hex), ('Rule', str)], self._generator())
|
||||
|
||||
@@ -49,6 +49,7 @@ class VerInfo(interfaces_plugins.PluginInterface):
|
||||
## TODO: we might add a regex option on the name later, but otherwise we're good
|
||||
## TODO: and we don't want any CLI options from pslist, modules, or moddump
|
||||
return [
|
||||
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (1, 0, 0)),
|
||||
requirements.TranslationLayerRequirement(
|
||||
name = 'primary', description = 'Memory layer for the kernel', architectures = ["Intel32", "Intel64"]),
|
||||
requirements.SymbolTableRequirement(name = "nt_symbols", description = "Windows kernel symbols"),
|
||||
@@ -155,8 +156,6 @@ class VerInfo(interfaces_plugins.PluginInterface):
|
||||
format_hints.Hex(entry.DllBase), BaseDllName, major, minor, product, build))
|
||||
|
||||
def run(self):
|
||||
self.check_plugin_version(pslist.PsList, (1, 0, 0))
|
||||
|
||||
procs = pslist.PsList.list_processes(self.context, self.config["primary"], self.config["nt_symbols"])
|
||||
|
||||
mods = modules.Modules.list_modules(self.context, self.config["primary"], self.config["nt_symbols"])
|
||||
|
||||
Reference in New Issue
Block a user