Improve docstrings for all plugins, and reformat all docstrings.

This commit is contained in:
Mike Auty
2019-09-07 22:59:54 +01:00
parent dc0a809729
commit e922cef316
134 changed files with 1933 additions and 1250 deletions
+4 -3
View File
@@ -1,9 +1,10 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""All core generic plugins
"""All core generic plugins.
These modules should only be imported from volatility.plugins NOT volatility.framework.plugins
These modules should only be imported from volatility.plugins NOT
volatility.framework.plugins
"""
import logging
@@ -19,7 +20,7 @@ def construct_plugin(context: interfaces.context.ContextInterface,
plugin: Type[interfaces.plugins.PluginInterface], base_config_path: str,
progress_callback: constants.ProgressCallback,
file_consumer: interfaces.plugins.FileConsumerInterface) -> interfaces.plugins.PluginInterface:
"""Constructs a plugin object based on the parameters
"""Constructs a plugin object based on the parameters.
Clever magic figures out how to fulfill each requirement that might not be fulfilled
+2 -1
View File
@@ -14,7 +14,8 @@ vollog = logging.getLogger(__name__)
class ConfigWriter(plugins.PluginInterface):
"""Runs the automagics and both prints and outputs configuration in the output directory"""
"""Runs the automagics and both prints and outputs configuration in the
output directory."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+2 -1
View File
@@ -14,7 +14,8 @@ vollog = logging.getLogger(__name__)
class LayerWriter(plugins.PluginInterface):
"""Runs the automagics and lists out the generated layers if no layer name is specified, otherwise writes out the named layer"""
"""Runs the automagics and lists out the generated layers if no layer name
is specified, otherwise writes out the named layer."""
default_output_name = "output.raw"
default_block_size = 0x500000
@@ -1,7 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""All core linux plugins
"""All core linux plugins.
These modules should only be imported from volatility.plugins NOT volatility.framework.plugins
These modules should only be imported from volatility.plugins NOT
volatility.framework.plugins
"""
+3 -4
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
import datetime
import struct
@@ -20,7 +19,7 @@ from volatility.plugins.linux import pslist
class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Recovers bash command history from memory"""
"""Recovers bash command history from memory."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
import logging
from typing import List
@@ -18,7 +17,7 @@ vollog = logging.getLogger(__name__)
class Check_afinfo(plugins.PluginInterface):
"""Verifies the operation function pointers of network protocols"""
"""Verifies the operation function pointers of network protocols."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
import logging
from typing import List
@@ -25,7 +24,7 @@ except ImportError:
class Check_syscall(plugins.PluginInterface):
"""Check system call table for hooks"""
"""Check system call table for hooks."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -36,9 +35,7 @@ class Check_syscall(plugins.PluginInterface):
]
def _get_table_size_next_symbol(self, table_addr, ptr_sz, vmlinux):
"""
Returns the size of the table based on the next symbol
"""
"""Returns the size of the table based on the next symbol."""
ret = 0
sym_table = self.context.symbol_space[vmlinux.name]
@@ -58,10 +55,9 @@ class Check_syscall(plugins.PluginInterface):
return ret
def _get_table_size_meta(self, vmlinux):
"""
returns the number of symbols that start with __syscall_meta__
this is a fast way to determine the number of system calls, but not the most accurate
"""
"""returns the number of symbols that start with __syscall_meta__ this
is a fast way to determine the number of system calls, but not the most
accurate."""
return len(
[sym for sym in self.context.symbol_space[vmlinux.name].symbols if sym.startswith("__syscall_meta__")])
@@ -77,11 +73,9 @@ class Check_syscall(plugins.PluginInterface):
return table_size
def _get_table_info_disassembly(self, ptr_sz, vmlinux):
"""
Find the size of the system call table by disassembling functions
that immediately reference it in their first isntruction
This is in the form 'cmp reg,NR_syscalls'
"""
"""Find the size of the system call table by disassembling functions
that immediately reference it in their first isntruction This is in the
form 'cmp reg,NR_syscalls'."""
table_size = 0
if not has_capstone:
+3 -4
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
from typing import List
@@ -16,7 +15,7 @@ from volatility.plugins.linux import pslist
class Elfs(plugins.PluginInterface):
"""Lists all memory mapped ELF files for all processes"""
"""Lists all memory mapped ELF files for all processes."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+16 -7
View File
@@ -1,11 +1,10 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
from typing import List
from typing import List, Generator, Iterable
from volatility.framework import contexts
from volatility.framework import renderers, constants, interfaces
@@ -17,7 +16,7 @@ from volatility.framework.renderers import format_hints
class Lsmod(plugins.PluginInterface):
"""Lists loaded kernel modules"""
"""Lists loaded kernel modules."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -28,8 +27,18 @@ class Lsmod(plugins.PluginInterface):
]
@classmethod
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str, vmlinux_symbols: str):
"""Lists all the modules in the primary layer"""
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str,
vmlinux_symbols: str) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the modules in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
vmlinux_symbols: The name of the table containing the kernel symbols
Yields:
The modules present in the `layer_name` layer's modules list
"""
linux.LinuxUtilities.aslr_mask_symbol_table(context, vmlinux_symbols, layer_name)
vmlinux = contexts.Module(context, vmlinux_symbols, layer_name, 0)
+3 -4
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
import logging
from typing import List
@@ -18,7 +17,7 @@ vollog = logging.getLogger(__name__)
class Lsof(plugins.PluginInterface):
"""Lists all memory maps for all processes"""
"""Lists all memory maps for all processes."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -15,7 +15,7 @@ from volatility.framework.renderers import format_hints
class Malfind(interfaces_plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code"""
"""Lists process memory ranges that potentially contain injected code."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -26,9 +26,8 @@ class Malfind(interfaces_plugins.PluginInterface):
]
def _list_injections(self, task):
"""Generate memory regions for a process that may contain
injected code.
"""
"""Generate memory regions for a process that may contain injected
code."""
proc_layer_name = task.add_process_layer()
if not proc_layer_name:
+3 -4
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Linux's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in Linux's /proc file system."""
from volatility.framework import renderers
from volatility.framework.configuration import requirements
@@ -14,7 +13,7 @@ from volatility.plugins.linux import pslist
class Maps(plugins.PluginInterface):
"""Lists all memory maps for all processes"""
"""Lists all memory maps for all processes."""
@classmethod
def get_requirements(cls):
+19 -2
View File
@@ -12,7 +12,7 @@ from volatility.framework.objects import utility
class PsList(interfaces_plugins.PluginInterface):
"""Lists the processes present in a particular linux memory image"""
"""Lists the processes present in a particular linux memory image."""
_version = (1, 0, 0)
@@ -26,6 +26,14 @@ class PsList(interfaces_plugins.PluginInterface):
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[Any], bool]:
"""Constructs a filter function for process IDs.
Args:
pid_list: List of process IDs that are acceptable (or None if all are acceptable)
Returns:
Function which, when provided a process object, returns True if the process is to be filtered out of the list
"""
# FIXME: mypy #4973 or #2608
pid_list = pid_list or []
filter_list = [x for x in pid_list if x is not None]
@@ -58,7 +66,16 @@ class PsList(interfaces_plugins.PluginInterface):
vmlinux_symbols: str,
filter_func: Callable[[int], bool] = lambda _: False
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the tasks in the primary layer"""
"""Lists all the tasks in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
vmlinux_symbols: The name of the table containing the kernel symbols
Yields:
Process objects
"""
linux.LinuxUtilities.aslr_mask_symbol_table(context, vmlinux_symbols, layer_name)
vmlinux = contexts.Module(context, vmlinux_symbols, layer_name, 0)
+4 -3
View File
@@ -7,7 +7,8 @@ from volatility.plugins.linux import pslist
class PsTree(pslist.PsList):
"""Plugin for listing processes in a tree based on their parent process ID """
"""Plugin for listing processes in a tree based on their parent process
ID."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -16,7 +17,7 @@ class PsTree(pslist.PsList):
self._children = {}
def find_level(self, pid):
"""Finds how deep the pid is in the processes list"""
"""Finds how deep the pid is in the processes list."""
seen = set([])
seen.add(pid)
level = 0
@@ -32,7 +33,7 @@ class PsTree(pslist.PsList):
self._levels[pid] = level
def _generator(self):
"""Generates the """
"""Generates the."""
for proc in self.list_tasks(self.context, self.config['primary'], self.config['vmlinux']):
self._processes[proc.pid] = proc
+3 -4
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in mac's /proc file system.
"""
"""A module containing a collection of plugins that produce data typically
found in mac's /proc file system."""
import datetime
import struct
@@ -20,7 +19,7 @@ from volatility.framework.symbols.linux.bash import BashIntermedSymbols
class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Recovers bash command history from memory"""
"""Recovers bash command history from memory."""
@classmethod
def get_requirements(cls):
@@ -15,7 +15,7 @@ vollog = logging.getLogger(__name__)
class Check_syscall(plugins.PluginInterface):
"""Check system call table for hooks"""
"""Check system call table for hooks."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -17,7 +17,7 @@ vollog = logging.getLogger(__name__)
class Check_sysctl(plugins.PluginInterface):
"""Check sysctl handlers for hooks"""
"""Check sysctl handlers for hooks."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -16,7 +16,7 @@ vollog = logging.getLogger(__name__)
class Check_trap_table(plugins.PluginInterface):
"""Check mach trap table for hooks"""
"""Check mach trap table for hooks."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+13 -5
View File
@@ -1,9 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""A module containing a collection of plugins that produce data
typically found in Mac's lsmod command.
"""
"""A module containing a collection of plugins that produce data typically
found in Mac's lsmod command."""
from volatility.framework import renderers, interfaces, contexts
from volatility.framework.automagic import mac
from volatility.framework.configuration import requirements
@@ -13,7 +12,7 @@ from volatility.framework.renderers import format_hints
class Lsmod(plugins.PluginInterface):
"""Lists loaded kernel modules"""
"""Lists loaded kernel modules."""
_version = (1, 0, 0)
@@ -27,7 +26,16 @@ class Lsmod(plugins.PluginInterface):
@classmethod
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str, darwin_symbols: str):
"""Lists all the modules in the primary layer"""
"""Lists all the modules in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
darwin_symbols: The name of the table containing the kernel symbols
Returns:
A list of modules from the `layer_name` layer
"""
mac.MacUtilities.aslr_mask_symbol_table(context, darwin_symbols, layer_name)
kernel = contexts.Module(context, darwin_symbols, layer_name, 0)
+1 -1
View File
@@ -14,7 +14,7 @@ vollog = logging.getLogger(__name__)
class lsof(plugins.PluginInterface):
"""Lists all open file descriptors for all processes"""
"""Lists all open file descriptors for all processes."""
@classmethod
def get_requirements(cls):
+3 -4
View File
@@ -13,7 +13,7 @@ from volatility.framework.renderers import format_hints
class Malfind(interfaces_plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code"""
"""Lists process memory ranges that potentially contain injected code."""
@classmethod
def get_requirements(cls):
@@ -24,9 +24,8 @@ class Malfind(interfaces_plugins.PluginInterface):
]
def _list_injections(self, task):
"""Generate memory regions for a process that may contain
injected code.
"""
"""Generate memory regions for a process that may contain injected
code."""
proc_layer_name = task.add_process_layer()
if proc_layer_name is None:
+1 -1
View File
@@ -16,7 +16,7 @@ vollog = logging.getLogger(__name__)
class Netstat(plugins.PluginInterface):
"""Lists all network connections for all processes"""
"""Lists all network connections for all processes."""
@classmethod
def get_requirements(cls):
@@ -12,7 +12,7 @@ from volatility.framework.renderers import format_hints
class Maps(interfaces_plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code"""
"""Lists process memory ranges that potentially contain injected code."""
@classmethod
def get_requirements(cls):
+2 -2
View File
@@ -1,7 +1,7 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""In-memory artifacts from OSX systems"""
"""In-memory artifacts from OSX systems."""
from typing import Iterator, Tuple, Any, Generator, List
from volatility.framework import exceptions, renderers, interfaces
@@ -12,7 +12,7 @@ from volatility.plugins.mac import pslist
class Psaux(plugins.PluginInterface):
"""Recovers program command line arguments"""
"""Recovers program command line arguments."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+12 -2
View File
@@ -14,7 +14,7 @@ vollog = logging.getLogger(__name__)
class PsList(interfaces.plugins.PluginInterface):
"""Lists the processes present in a particular mac memory image"""
"""Lists the processes present in a particular mac memory image."""
_version = (1, 0, 0)
@@ -59,7 +59,17 @@ class PsList(interfaces.plugins.PluginInterface):
darwin_symbols: str,
filter_func: Callable[[int], bool] = lambda _: False) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the tasks in the primary layer"""
"""Lists all the processes in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
darwin_symbols: The name of the table containing the kernel symbols
filter_func: A function which takes a process object and returns True if the process should be ignored/filtered
Returns:
The list of process objects from the processes linked list after filtering
"""
mac.MacUtilities.aslr_mask_symbol_table(context, darwin_symbols, layer_name)
+4 -3
View File
@@ -10,7 +10,8 @@ from volatility.plugins.mac import pslist
class PsTree(plugins.PluginInterface):
"""Plugin for listing processes in a tree based on their parent process ID """
"""Plugin for listing processes in a tree based on their parent process
ID."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -28,7 +29,7 @@ class PsTree(plugins.PluginInterface):
]
def _find_level(self, pid):
"""Finds how deep the pid is in the processes list"""
"""Finds how deep the pid is in the processes list."""
seen = set([])
seen.add(pid)
level = 0
@@ -43,7 +44,7 @@ class PsTree(plugins.PluginInterface):
self._levels[pid] = level
def _generator(self):
"""Generates the """
"""Generates the."""
for proc in pslist.PsList.list_tasks(self.context, self.config['primary'], self.config['darwin']):
self._processes[proc.p_pid] = proc
+12 -2
View File
@@ -13,7 +13,7 @@ vollog = logging.getLogger(__name__)
class Tasks(pslist.PsList):
"""Lists the processes present in a particular mac memory image"""
"""Lists the processes present in a particular mac memory image."""
@classmethod
def list_tasks(cls,
@@ -22,7 +22,17 @@ class Tasks(pslist.PsList):
darwin_symbols: str,
filter_func: Callable[[int], bool] = lambda _: False) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the tasks in the primary layer"""
"""Lists all the tasks in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
darwin_symbols: The name of the table containing the kernel symbols
filter_func: A function which takes a task object and returns True if the task should be ignored/filtered
Returns:
The list of task objects from the `layer_name` layer's `tasks` list after filtering
"""
mac.MacUtilities.aslr_mask_symbol_table(context, darwin_symbols, layer_name)
@@ -18,7 +18,7 @@ vollog = logging.getLogger(__name__)
class Check_syscall(plugins.PluginInterface):
"""Check system call table for hooks"""
"""Check system call table for hooks."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+11 -6
View File
@@ -26,18 +26,21 @@ class TimeLinerType(enum.IntEnum):
class TimeLinerInterface(metaclass = abc.ABCMeta):
"""Interface defining methods that timeliner will use to generate a body file"""
"""Interface defining methods that timeliner will use to generate a body
file."""
@abc.abstractmethod
def generate_timeline(self) -> Generator[Tuple[str, TimeLinerType, datetime.datetime], None, None]:
"""Method generates Tuples of (description, timestamp_type, timestamp)
These need not be generated in any particular order, sorting will be done later
These need not be generated in any particular order, sorting
will be done later
"""
class Timeliner(interfaces.plugins.PluginInterface):
"""Runs all relevant plugins that provide time related information and orders the results by time"""
"""Runs all relevant plugins that provide time related information and
orders the results by time."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -82,7 +85,8 @@ class Timeliner(interfaces.plugins.PluginInterface):
]
def _generator(self, runable_plugins: List[TimeLinerInterface]) -> Optional[Iterable[Tuple[int, Tuple]]]:
"""Takes a timeline, sorts it and output the data from each relevant row from each plugin"""
"""Takes a timeline, sorts it and output the data from each relevant
row from each plugin."""
# Generate the results for each plugin
for plugin in runable_plugins:
plugin_name = plugin.__class__.__name__
@@ -112,7 +116,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
yield data
def run(self):
"""Isolate each plugin and run it"""
"""Isolate each plugin and run it."""
# Use all the plugins if there's no filter
self.usable_plugins = self.usable_plugins or self.get_usable_plugins()
@@ -153,6 +157,7 @@ class Timeliner(interfaces.plugins.PluginInterface):
generator = self._generator(runable_plugins))
def build_configuration(self):
"""Builds the configuration to save for the plugin such that it can be reconstructed"""
"""Builds the configuration to save for the plugin such that it can be
reconstructed."""
vollog.warning("Unable to record configuration data for the timeliner plugin")
return []
@@ -1,7 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""All core windows plugins
"""All core windows plugins.
These modules should only be imported from volatility.plugins NOT volatility.framework.plugins
These modules should only be imported from volatility.plugins NOT
volatility.framework.plugins
"""
@@ -12,7 +12,7 @@ from volatility.plugins.windows import pslist
class CmdLine(interfaces_plugins.PluginInterface):
"""Lists process command line arguments"""
"""Lists process command line arguments."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -22,7 +22,7 @@ vollog = logging.getLogger(__name__)
class DllDump(interfaces_plugins.PluginInterface):
"""Dumps process memory ranges as DLLs"""
"""Dumps process memory ranges as DLLs."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -12,7 +12,7 @@ from volatility.plugins.windows import pslist
class DllList(interfaces_plugins.PluginInterface):
"""Lists the loaded modules in a particular windows memory image"""
"""Lists the loaded modules in a particular windows memory image."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -21,7 +21,7 @@ MAJOR_FUNCTIONS = [
class DriverIrp(plugins.PluginInterface):
"""List IRPs for drivers in a particular windows memory image"""
"""List IRPs for drivers in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -13,7 +13,7 @@ from volatility.framework.renderers import format_hints
class DriverScan(plugins.PluginInterface):
"""Scans for drivers present in a particular windows memory image"""
"""Scans for drivers present in a particular windows memory image."""
_version = (1, 0, 0)
@@ -31,7 +31,16 @@ class DriverScan(plugins.PluginInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for drivers using the poolscanner module and constraints"""
"""Scans for drivers using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of Driver objects as found from the `layer_name` layer based on Driver pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Dri\xf6', b'Driv'])
@@ -4,15 +4,16 @@
from typing import Iterable
import volatility.plugins.windows.poolscanner as poolscanner
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import renderers, interfaces, exceptions
from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class FileScan(plugins.PluginInterface):
"""Scans for file objects present in a particular windows memory image"""
"""Scans for file objects present in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -28,7 +29,16 @@ class FileScan(plugins.PluginInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for file objects using the poolscanner module and constraints"""
"""Scans for file objects using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of File objects as found from the `layer_name` layer based on File pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Fil\xe5', b'File'])
+35 -19
View File
@@ -3,7 +3,7 @@
#
import logging
from typing import List, Optional
from typing import List, Optional, Dict
import volatility.plugins.windows.pslist as pslist
@@ -24,7 +24,7 @@ except ImportError:
class Handles(interfaces_plugins.PluginInterface):
"""Lists process open handles"""
"""Lists process open handles."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -46,8 +46,11 @@ class Handles(interfaces_plugins.PluginInterface):
def _decode_pointer(self, value, magic):
"""Windows encodes pointers to objects and decodes them on the fly
before using them. This function mimics the decoding routine so we
can generate the proper pointer values as well."""
before using them.
This function mimics the decoding routine so we can generate the
proper pointer values as well.
"""
value = value & 0xFFFFFFFFFFFFFFF8
value = value >> magic
@@ -57,8 +60,8 @@ class Handles(interfaces_plugins.PluginInterface):
return value
def _get_item(self, handle_table_entry, handle_value):
"""Given a handle table entry (_HANDLE_TABLE_ENTRY) structure from
a process' handle table, determine where the corresponding object's
"""Given a handle table entry (_HANDLE_TABLE_ENTRY) structure from a
process' handle table, determine where the corresponding object's
_OBJECT_HEADER can be found."""
virtual = self.config["primary"]
@@ -91,10 +94,12 @@ class Handles(interfaces_plugins.PluginInterface):
return object_header
def find_sar_value(self):
"""Locate ObpCaptureHandleInformationEx if it exists in the
sample. Once found, parse it for the SAR value that we need
to decode pointers in the _HANDLE_TABLE_ENTRY which allows us
to find the associated _OBJECT_HEADER."""
"""Locate ObpCaptureHandleInformationEx if it exists in the sample.
Once found, parse it for the SAR value that we need to decode
pointers in the _HANDLE_TABLE_ENTRY which allows us to find the
associated _OBJECT_HEADER.
"""
if self._sar_value is None:
@@ -128,14 +133,25 @@ class Handles(interfaces_plugins.PluginInterface):
return self._sar_value
@classmethod
def list_objects(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str) -> dict:
def list_objects(cls, context: interfaces.context.ContextInterface, layer_name: str,
symbol_table: str) -> Dict[int, str]:
"""List the executive object types (_OBJECT_TYPE) using the
ObTypeIndexTable or ObpObjectTypes symbol (differs per OS).
This method will be necessary for determining what type of
object we have given an object header.
ObTypeIndexTable or ObpObjectTypes symbol (differs per OS). This method
will be necessary for determining what type of object we have given an
object header.
Note: The object type index map was hard coded into profiles
in vol2, but we generate it dynamically now."""
Note:
The object type index map was hard coded into profiles in previous versions of volatility.
It is now generated dynamically.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A mapping of type indicies to type names
"""
type_map = {}
@@ -150,7 +166,7 @@ class Handles(interfaces_plugins.PluginInterface):
ptrs = ntkrnlmp.object(
object_type = "array", offset = table_addr, subtype = ntkrnlmp.get_type("pointer"), count = 100)
for i, ptr in enumerate(ptrs): #type: ignore
for i, ptr in enumerate(ptrs): # type: ignore
# the first entry in the table is always null. break the
# loop when we encounter the first null entry after that
if i > 0 and ptr == 0:
@@ -182,8 +198,8 @@ class Handles(interfaces_plugins.PluginInterface):
return context.object(symbol_table + constants.BANG + "unsigned int", layer_name, offset = kvo + offset)
def _make_handle_array(self, offset, level, depth = 0):
"""Parse a process' handle table and yield valid handle table
entries, going as deep into the table "levels" as necessary."""
"""Parse a process' handle table and yield valid handle table entries,
going as deep into the table "levels" as necessary."""
virtual = self.config["primary"]
kvo = self.context.layers[virtual].config['kernel_virtual_offset']
+3 -3
View File
@@ -3,11 +3,11 @@
#
import time
from typing import List
from typing import List, Tuple, Iterable
from volatility.framework.interfaces import plugins
from volatility.framework import constants, interfaces, layers
from volatility.framework.configuration import requirements
from volatility.framework.interfaces import plugins
from volatility.framework.renderers import TreeGrid
from volatility.framework.symbols import intermed
from volatility.framework.symbols.windows import extensions
@@ -15,7 +15,7 @@ from volatility.framework.symbols.windows.extensions import kdbg
class Info(plugins.PluginInterface):
"""Show OS & kernel details of the memory sample being analyzed"""
"""Show OS & kernel details of the memory sample being analyzed."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+19 -10
View File
@@ -1,6 +1,7 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
from typing import Iterable, Tuple
import volatility.plugins.windows.pslist as pslist
import volatility.plugins.windows.vadinfo as vadinfo
@@ -13,7 +14,7 @@ from volatility.framework.renderers import format_hints
class Malfind(interfaces.plugins.PluginInterface):
"""Lists process memory ranges that potentially contain injected code"""
"""Lists process memory ranges that potentially contain injected code."""
@classmethod
def get_requirements(cls):
@@ -27,16 +28,18 @@ class Malfind(interfaces.plugins.PluginInterface):
]
@classmethod
def is_vad_empty(self, proc_layer, vad):
"""Check if a VAD region is either entirely unavailable
due to paging, entirely consisting of zeros, or a
combination of the two. This helps ignore false positives
whose VAD flags match task._injection_filter requirements
but there's no data and thus not worth reporting it.
def is_vad_empty(cls, proc_layer, vad):
"""Check if a VAD region is either entirely unavailable due to paging,
entirely consisting of zeros, or a combination of the two. This helps
ignore false positives whose VAD flags match task._injection_filter
requirements but there's no data and thus not worth reporting it.
Args:
proc_layer: the process layer
vad: the MMVAD structure to test
Returns:
A boolean indicating whether a vad is empty or not
"""
CHUNK_SIZE = 0x1000
@@ -55,12 +58,18 @@ class Malfind(interfaces.plugins.PluginInterface):
@classmethod
def list_injections(cls, context: interfaces.context.ContextInterface, symbol_table: str,
proc: interfaces.objects.ObjectInterface):
"""Generate memory regions for a process that may contain
injected code.
proc: interfaces.objects.ObjectInterface
) -> Iterable[Tuple[interfaces.objects.ObjectInterface, bytes]]:
"""Generate memory regions for a process that may contain injected
code.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
symbol_table: The name of the table containing the kernel symbols
proc: an _EPROCESS instance
Returns:
An iterable of VAD instances and the first 64 bytes of data containing in that region
"""
proc_layer_name = proc.add_process_layer()
@@ -18,7 +18,7 @@ vollog = logging.getLogger(__name__)
class ModDump(interfaces.plugins.PluginInterface):
"""Dumps kernel modules"""
"""Dumps kernel modules."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -41,8 +41,14 @@ class ModDump(interfaces.plugins.PluginInterface):
the primary/kernel layer. Then keep one layer per session by cycling
through the process list.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
pids: A list of process identifiers to include exclusively or None for no filter
Returns:
<list> of layer names
A list of session layer names
"""
seen_ids = [] # type: List[interfaces.objects.ObjectInterface]
filter_func = pslist.PsList.create_pid_filter(pids or [])
@@ -72,15 +78,18 @@ class ModDump(interfaces.plugins.PluginInterface):
@classmethod
def find_session_layer(cls, context: interfaces.context.ContextInterface, session_layers: Iterable[str],
base_address: int):
"""Given a base address and a list of layer names, find a
layer that can access the specified address.
"""Given a base address and a list of layer names, find a layer that
can access the specified address.
Args:
session_layers: <list> of layer names
base_address: <int> the base address
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
session_layers: A list of session layer names
base_address: The base address to identify the layers that can access it
Returns:
layer name (or None)
Layer name or None if no layers that contain the base address can be found
"""
for layer_name in session_layers:
@@ -13,7 +13,7 @@ from volatility.framework.renderers import format_hints
class ModScan(plugins.PluginInterface):
"""Scans for modules present in a particular windows memory image"""
"""Scans for modules present in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -29,7 +29,16 @@ class ModScan(plugins.PluginInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for modules using the poolscanner module and constraints"""
"""Scans for modules using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of Driver objects as found from the `layer_name` layer based on Driver pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'MmLd'])
@@ -2,7 +2,7 @@
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
from typing import List
from typing import List, Iterable
from volatility.framework import constants
from volatility.framework import exceptions, interfaces
@@ -12,7 +12,7 @@ from volatility.framework.renderers import format_hints
class Modules(interfaces.plugins.PluginInterface):
"""Lists the loaded kernel modules"""
"""Lists the loaded kernel modules."""
_version = (1, 0, 0)
@@ -46,8 +46,18 @@ class Modules(interfaces.plugins.PluginInterface):
))
@classmethod
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str, symbol_table: str):
"""Lists all the modules in the primary layer"""
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str,
symbol_table: str) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the modules in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of Modules as retrieved from PsLoadedModuleList
"""
kvo = context.layers[layer_name].config['kernel_virtual_offset']
ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
@@ -4,15 +4,16 @@
from typing import Iterable
import volatility.plugins.windows.poolscanner as poolscanner
import volatility.framework.interfaces.plugins as plugins
from volatility.framework import renderers, interfaces, exceptions
from volatility.framework.configuration import requirements
from volatility.framework.renderers import format_hints
import volatility.plugins.windows.poolscanner as poolscanner
class MutantScan(plugins.PluginInterface):
"""Scans for mutexes present in a particular windows memory image"""
"""Scans for mutexes present in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -28,7 +29,16 @@ class MutantScan(plugins.PluginInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for mutants using the poolscanner module and constraints"""
"""Scans for mutants using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of Mutant objects found by scanning memory for the Mutant pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Mut\xe1', b'Muta'])
@@ -22,8 +22,8 @@ vollog = logging.getLogger(__name__)
# TODO: When python3.5 is no longer supported, make this enum.IntFlag
class PoolType(enum.IntEnum):
"""Class to maintain the different possible PoolTypes
The values must be integer powers of 2"""
"""Class to maintain the different possible PoolTypes The values must be
integer powers of 2."""
PAGED = 1
NONPAGED = 2
@@ -31,7 +31,8 @@ class PoolType(enum.IntEnum):
class PoolConstraint:
"""Class to maintain tag/size/index/type information about Pool header tags"""
"""Class to maintain tag/size/index/type information about Pool header
tags."""
def __init__(self,
tag: bytes,
@@ -110,24 +111,42 @@ class PoolHeaderScanner(interfaces.layers.ScannerInterface):
def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
fallback_checks: List[Tuple[str, Optional[str], bool]]
) -> Callable[[interfaces.context.ContextInterface, str], bool]:
"""Distinguishes a symbol table as being above a particular version or point
"""Distinguishes a symbol table as being above a particular version or
point.
This will primarily check the version metadata first and foremost.
If that metadata isn't available then each item in the fallback_checks is tested.
If invert is specified then the result will be true if the version is less than that specified, or in the case of
fallback, if any of the fallback checks is successful.
This will primarily check the version metadata first and foremost.
If that metadata isn't available then each item in the fallback_checks is tested.
If invert is specified then the result will be true if the version is less than that specified, or in the case of
fallback, if any of the fallback checks is successful.
A fallback check is made up of:
* a symbol or type name
* a member name (implying that the value before was a type name)
* whether that symbol, type or member must be present or absent for the symbol table to be more above the required point
A fallback check is made up of:
* a symbol or type name
* a member name (implying that the value before was a type name)
* whether that symbol, type or member must be present or absent for the symbol table to be more above the required point
Note: Specifying that a member must not be present includes the whole type not being present too (ie, either will pass the test)
Note:
Specifying that a member must not be present includes the whole type not being present too (ie, either will pass the test)
Args:
version_check: Function that takes a 4-tuple version and returns whether whether the provided version is above a particular point
fallback_checks: A list of symbol/types/members of types, and whether they must be present to be above the required point
Returns:
A function that takes a context and a symbol table name and determines whether that symbol table passes the distinguishing checks
"""
# try the primary method based on the pe version in the ISF
@functools.wraps(version_check)
def method(context: interfaces.context.ContextInterface, symbol_table: str) -> bool:
"""
Args:
context: The context that contains the symbol table named `symbol_table`
symbol_table: Name of the symbol table within the context to distinguish the version of
Returns:
True if the symbol table is of the required version
"""
try:
pe_version = context.symbol_space[symbol_table].metadata.pe_version
@@ -160,7 +179,7 @@ def os_distinguisher(version_check: Callable[[Tuple[int, ...]], bool],
class PoolScanner(plugins.PluginInterface):
"""A generic pool scanner plugin"""
"""A generic pool scanner plugin."""
_version = (1, 0, 0)
@@ -210,6 +229,13 @@ class PoolScanner(plugins.PluginInterface):
The tags_filter is a list of pool tags, and the associated
PoolConstraints are returned. If tags_filter is empty or
not supplied, then all builtin constraints are returned.
Args:
symbol_table: The name of the symbol table to prepend to the types used
tags_filter: List of tags to return or None to return all
Returns:
A list of well-known constructed PoolConstraints that match the provided tags
"""
builtins = [
@@ -316,6 +342,17 @@ class PoolScanner(plugins.PluginInterface):
constraints: List[PoolConstraint]) \
-> Generator[Tuple[
PoolConstraint, interfaces.objects.ObjectInterface, interfaces.objects.ObjectInterface], None, None]:
"""
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
constraints: List of pool constraints used to limit the scan results
Returns:
Iterable of tuples, containing the constraint that matched, the object from memory, the object header used to determine the object
"""
# get the object type map
type_map = handles.Handles.list_objects(context = context, layer_name = layer_name, symbol_table = symbol_table)
@@ -357,8 +394,22 @@ class PoolScanner(plugins.PluginInterface):
alignment: int = 8,
progress_callback: Optional[constants.ProgressCallback] = None) \
-> Generator[Tuple[PoolConstraint, interfaces.objects.ObjectInterface], None, None]:
"""Returns the _POOL_HEADER object (based on the symbol_table template) after scanning through layer_name
returning all headers that match any of the constraints provided. Only one constraint can be provided per tag"""
"""Returns the _POOL_HEADER object (based on the symbol_table template)
after scanning through layer_name returning all headers that match any
of the constraints provided. Only one constraint can be provided per
tag.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
pool_constraints: List of pool constraints used to limit the scan results
alignment: An optional value that all pool headers will be aligned to
progress_callback: An optional function to provide progress feedback whilst scanning
Returns:
An Iterable of pool constraints and the pool headers associated with them
"""
# Setup the pattern
constraint_lookup = {} # type: Dict[bytes, PoolConstraint]
for constraint in pool_constraints:
@@ -21,7 +21,7 @@ vollog = logging.getLogger(__name__)
class ProcDump(interfaces_plugins.PluginInterface):
"""Dumps process executable images"""
"""Dumps process executable images."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
+31 -2
View File
@@ -14,7 +14,7 @@ from volatility.plugins import timeliner
class PsList(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Lists the processes present in a particular windows memory image"""
"""Lists the processes present in a particular windows memory image."""
_version = (1, 0, 0)
PHYSICAL_DEFAULT = False
@@ -37,6 +37,15 @@ class PsList(plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def create_pid_filter(cls, pid_list: List[int] = None) -> Callable[[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process IDs.
Args:
pid_list: A list of process IDs that are acceptable, all other processes will be filtered out
Returns:
Filter function for passing to the `list_processes` method
"""
filter_func = lambda _: False
# FIXME: mypy #4973 or #2608
pid_list = pid_list or []
@@ -47,6 +56,15 @@ class PsList(plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def create_name_filter(cls, name_list: List[str] = None) -> Callable[[interfaces.objects.ObjectInterface], bool]:
"""A factory for producing filter functions that filter based on a list
of process names.
Args:
name_list: A list of process names that are acceptable, all other processes will be filtered out
Returns:
Filter function for passing to the `list_processes` method
"""
filter_func = lambda _: False
# FIXME: mypy #4973 or #2608
name_list = name_list or []
@@ -62,7 +80,18 @@ class PsList(plugins.PluginInterface, timeliner.TimeLinerInterface):
symbol_table: str,
filter_func: Callable[[interfaces.objects.ObjectInterface], bool] = lambda _: False) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the processes in the primary layer that are in the pid config option"""
"""Lists all the processes in the primary layer that are in the pid
config option.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
filter_func: A function which takes an EPROCESS object and returns True if the process should be ignored/filtered
Returns:
The list of EPROCESS objects from the `layer_name` layer's PsActiveProcessHead list after filtering
"""
# We only use the object factory to demonstrate how to use one
kvo = context.layers[layer_name].config['kernel_virtual_offset']
+11 -2
View File
@@ -14,7 +14,7 @@ import volatility.plugins.windows.poolscanner as poolscanner
class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for processes present in a particular windows memory image"""
"""Scans for processes present in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -30,7 +30,16 @@ class PsScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for processes using the poolscanner module and constraints"""
"""Scans for processes using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of processes found by scanning the `layer_name` layer for process pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Pro\xe3', b'Proc'])
@@ -10,7 +10,8 @@ from volatility.plugins.windows import pslist
class PsTree(pslist.PsList):
"""Plugin for listing processes in a tree based on their parent process ID """
"""Plugin for listing processes in a tree based on their parent process
ID."""
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
@@ -19,7 +20,7 @@ class PsTree(pslist.PsList):
self._children = {} # type: Dict[int, Set[int]]
def find_level(self, pid: objects.Pointer) -> None:
"""Finds how deep the pid is in the processes list"""
"""Finds how deep the pid is in the processes list."""
seen = set([])
seen.add(pid)
level = 0
@@ -33,7 +34,7 @@ class PsTree(pslist.PsList):
self._levels[pid] = level
def _generator(self):
"""Generates the Tree of processes"""
"""Generates the Tree of processes."""
for proc in self.list_processes(self.context, self.config['primary'], self.config['nt_symbols']):
if not self.config.get('physical', self.PHYSICAL_DEFAULT):
@@ -1,7 +1,8 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl_v1.0
#
"""All core windows registry plugins
"""All core windows registry plugins.
These modules should only be imported from volatility.plugins NOT volatility.framework.plugins
These modules should only be imported from volatility.plugins NOT
volatility.framework.plugins
"""
@@ -13,7 +13,7 @@ vollog = logging.getLogger(__name__)
class HiveList(plugins.PluginInterface):
"""Lists the registry hives present in a particular memory image"""
"""Lists the registry hives present in a particular memory image."""
_version = (1, 0, 0)
@@ -41,8 +41,18 @@ class HiveList(plugins.PluginInterface):
context: interfaces.context.ContextInterface,
layer_name: str,
symbol_table: str,
filter_string: None = None) -> Iterator[interfaces.objects.ObjectInterface]:
"""Lists all the hives in the primary layer"""
filter_string: str = None) -> Iterator[interfaces.objects.ObjectInterface]:
"""Lists all the hives in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
filter_string: A string which must be present in the hive name if specified
Returns:
The list of registry hives from the `layer_name` layer as filtered against using the `filter_string`
"""
# We only use the object factory to demonstrate how to use one
kvo = context.layers[layer_name].config['kernel_virtual_offset']
@@ -13,7 +13,8 @@ from volatility.framework.renderers import format_hints
class HiveScan(plugins.PluginInterface):
"""Scans for registry hives present in a particular windows memory image"""
"""Scans for registry hives present in a particular windows memory
image."""
@classmethod
def get_requirements(cls):
@@ -29,7 +30,16 @@ class HiveScan(plugins.PluginInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for hives using the poolscanner module and constraints"""
"""Scans for hives using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of Hive objects as found from the `layer_name` layer based on Hive pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'CM10'])
@@ -17,7 +17,7 @@ vollog = logging.getLogger(__name__)
class PrintKey(interfaces.plugins.PluginInterface):
"""Lists the registry keys under a hive or specific key value"""
"""Lists the registry keys under a hive or specific key value."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -36,8 +36,17 @@ class PrintKey(interfaces.plugins.PluginInterface):
@classmethod
def hive_walker(cls, hive: RegistryHive, node_path: Sequence[objects.StructType] = None,
recurse: bool = False) -> Generator:
"""Walks through a set of nodes from a given node (last one in node_path).
Avoids loops by not traversing into nodes already present in the node_path
"""Walks through a set of nodes from a given node (last one in
node_path). Avoids loops by not traversing into nodes already present
in the node_path.
Args:
hive: The registry hive to walk
node_path: The list of nodes that make up the
recurse: Traverse down the node tree or stay only on the same level
Yields:
The depth, and a tuple of results (last write time, hive offset, type, path, name, data and volatile)
"""
if not node_path:
node_path = [hive.get_node(hive.root_cell_offset)]
@@ -100,7 +109,7 @@ class PrintKey(interfaces.plugins.PluginInterface):
symbol_table: str,
offset: int = None,
key: str = None):
"""Walks through a registry, hive by hive"""
"""Walks through a registry, hive by hive."""
if offset is None:
try:
hive_offsets = [
@@ -20,7 +20,7 @@ vollog = logging.getLogger(__name__)
class UserAssist(interfaces.plugins.PluginInterface):
"""Print userassist registry keys and information"""
"""Print userassist registry keys and information."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -41,7 +41,8 @@ class UserAssist(interfaces.plugins.PluginInterface):
]
def parse_userassist_data(self, reg_val):
"""Reads the raw data of a _CM_KEY_VALUE and returns a dict of userassist fields"""
"""Reads the raw data of a _CM_KEY_VALUE and returns a dict of
userassist fields."""
item = {
"id": renderers.UnparsableValue(),
@@ -95,7 +96,8 @@ class UserAssist(interfaces.plugins.PluginInterface):
return item
def _determine_userassist_type(self) -> None:
"""Determine the userassist type and size depending on the OS version"""
"""Determine the userassist type and size depending on the OS
version."""
if self._win7 is True:
self._userassist_type_name = "_VOL_USERASSIST_TYPES_7"
+11 -2
View File
@@ -17,7 +17,7 @@ from volatility.plugins.windows import modules
class SSDT(plugins.PluginInterface):
"""Lists the system call table"""
"""Lists the system call table."""
_version = (1, 0, 0)
@@ -33,7 +33,16 @@ class SSDT(plugins.PluginInterface):
@classmethod
def build_module_collection(cls, context: interfaces.context.ContextInterface, layer_name: str,
symbol_table: str) -> contexts.ModuleCollection:
"""Builds a collection of modules"""
"""Builds a collection of modules.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A Module collection of available modules based on `Modules.list_modules`
"""
mods = modules.Modules.list_modules(context, layer_name, symbol_table)
context_modules = []
@@ -33,7 +33,7 @@ class Strings(interfaces.plugins.PluginInterface):
self._generator())
def _generator(self) -> Generator[Tuple, None, None]:
"""Generates results from a strings file"""
"""Generates results from a strings file."""
revmap = self.generate_mapping(self.config['primary'])
accessor = resources.ResourceAccessor()
@@ -52,7 +52,14 @@ class Strings(interfaces.plugins.PluginInterface):
@staticmethod
def _parse_line(line: bytes) -> Tuple[int, bytes]:
"""Parses a single line from a strings file"""
"""Parses a single line from a strings file.
Args:
line: bytes of the line of a strings file (an offset and a string)
Returns:
Tuple of the offset and the string found at that offset
"""
pattern = re.compile(rb"(?:\W*)([0-9]+)(?:\W*)(\w[\w\W]+)")
match = pattern.search(line)
if not match:
@@ -61,7 +68,15 @@ class Strings(interfaces.plugins.PluginInterface):
return int(offset), string
def generate_mapping(self, layer_name: str) -> Dict[int, Set[Tuple[str, int]]]:
"""Creates a reverse mapping between virtual addresses and physical addresses"""
"""Creates a reverse mapping between virtual addresses and physical
addresses.
Args:
layer_name: the layer to map against the string lines
Returns:
A mapping of virtual offsets to strings and physical offsets
"""
layer = self._context.layers[layer_name]
reverse_map = dict() # type: Dict[int, Set[Tuple[str, int]]]
if isinstance(layer, intel.Intel):
@@ -17,7 +17,7 @@ vollog = logging.getLogger(__name__)
class SvcScan(interfaces.plugins.PluginInterface):
"""Scans for windows services"""
"""Scans for windows services."""
is_vista_or_later = poolscanner.os_distinguisher(
version_check = lambda x: x >= (6, 0), fallback_checks = [("KdCopyDataBlock", None, True)])
@@ -60,7 +60,17 @@ class SvcScan(interfaces.plugins.PluginInterface):
@staticmethod
def create_service_table(context: interfaces.context.ContextInterface, symbol_table: str, config_path: str) -> str:
"""Constructs a symbol table containing the symbols for services
depending upon the operating system in use.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
symbol_table: The name of the table containing the kernel symbols
config_path: The configuration path for any settings required by the new table
Returns:
A symbol table containing the symbols necessary for services
"""
native_types = context.symbol_space[symbol_table].natives
is_64bit = symbols.symbol_table_is_64bit(context, symbol_table)
@@ -13,7 +13,7 @@ from volatility.plugins import timeliner
class SymlinkScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
"""Scans for links present in a particular windows memory image"""
"""Scans for links present in a particular windows memory image."""
@classmethod
def get_requirements(cls):
@@ -29,7 +29,16 @@ class SymlinkScan(plugins.PluginInterface, timeliner.TimeLinerInterface):
layer_name: str,
symbol_table: str) -> \
Iterable[interfaces.objects.ObjectInterface]:
"""Scans for links using the poolscanner module and constraints"""
"""Scans for links using the poolscanner module and constraints.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
Returns:
A list of symlink objects found by scanning memory for the Symlink pool signatures
"""
constraints = poolscanner.PoolScanner.builtin_constraints(symbol_table, [b'Sym\xe2', b'Symb'])
@@ -17,7 +17,7 @@ vollog = logging.getLogger(__name__)
class VadDump(interfaces_plugins.PluginInterface):
"""Dumps process memory ranges"""
"""Dumps process memory ranges."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -32,7 +32,7 @@ winnt_protections = {
class VadInfo(interfaces.plugins.PluginInterface):
"""Lists process memory ranges"""
"""Lists process memory ranges."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -57,23 +57,38 @@ class VadInfo(interfaces.plugins.PluginInterface):
]
@classmethod
def protect_values(cls, context: interfaces.context.ContextInterface, virtual_layer: str,
nt_symbols: str) -> Iterable[int]:
"""Look up the array of memory protection constants from the memory sample.
These don't change often, but if they do in the future, then finding them
# dynamically versus hard-coding here will ensure we parse them properly."""
def protect_values(cls, context: interfaces.context.ContextInterface, layer_name: str,
symbol_table: str) -> Iterable[int]:
"""Look up the array of memory protection constants from the memory
sample. These don't change often, but if they do in the future, then
finding them dynamically versus hard-coding here will ensure we parse
them properly.
kvo = context.layers[virtual_layer].config["kernel_virtual_offset"]
ntkrnlmp = context.module(nt_symbols, layer_name = virtual_layer, offset = kvo)
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
symbol_table: The name of the table containing the kernel symbols
"""
kvo = context.layers[layer_name].config["kernel_virtual_offset"]
ntkrnlmp = context.module(symbol_table, layer_name = layer_name, offset = kvo)
addr = ntkrnlmp.get_symbol("MmProtectToValue").address
values = ntkrnlmp.object(object_type = "array", offset = addr, subtype = ntkrnlmp.get_type("int"), count = 32)
return values # type: ignore
@classmethod
def list_vads(cls, proc: interfaces.objects.ObjectInterface,
filter_func: Callable[[int], bool] = lambda _: False) -> \
filter_func: Callable[[interfaces.objects.ObjectInterface], bool] = lambda _: False) -> \
Generator[interfaces.objects.ObjectInterface, None, None]:
"""Lists the Virtual Address Descriptors of a specific process.
Args:
proc: _EPROCESS object from which to list the VADs
filter_func: Function to take a virtual address descriptor value and return True if it should be filtered out
Returns:
A list of virtual address descriptors based on the process and filtered based on the filter function
"""
for vad in proc.get_vad_root().traverse():
if not filter_func(vad):
yield vad
@@ -77,7 +77,15 @@ class VadYaraScan(interfaces.plugins.PluginInterface):
@staticmethod
def get_vad_maps(task: interfaces.objects.ObjectInterface) -> Iterable[Tuple[int, int]]:
"""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
"""
vad_root = task.get_vad_root()
for vad in vad_root.traverse():
end = vad.get_end()
@@ -26,7 +26,7 @@ except ImportError:
class VerInfo(interfaces_plugins.PluginInterface):
"""Lists version information from PE files"""
"""Lists version information from PE files."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -42,7 +42,7 @@ class VerInfo(interfaces_plugins.PluginInterface):
@classmethod
def get_version_information(cls, context: interfaces.context.ContextInterface, pe_table_name: str, layer_name: str,
base_address: int) -> Tuple[int, int, int, int]:
"""Get File and Product version information from PE files
"""Get File and Product version information from PE files.
Args:
context: volatility context on which to operate
@@ -85,7 +85,8 @@ class VerInfo(interfaces_plugins.PluginInterface):
def _generator(self, procs: Generator[interfaces.objects.ObjectInterface, None, None],
mods: Generator[interfaces.objects.ObjectInterface, None, None],
session_layers: Generator[str, None, None]):
"""Generates a list of PE file version info for processes, dlls, and modules.
"""Generates a list of PE file version info for processes, dlls, and
modules.
Args:
procs: <generator> of processes
@@ -13,7 +13,7 @@ vollog = logging.getLogger(__name__)
class VirtMap(interfaces.plugins.PluginInterface):
"""Lists virtual mapped sections"""
"""Lists virtual mapped sections."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -35,7 +35,7 @@ class VirtMap(interfaces.plugins.PluginInterface):
@classmethod
def determine_map(cls, module: interfaces.context.ModuleInterface) -> \
Dict[int, List[Tuple[int, int]]]:
"""Returns the virtual map from a windows kernel module"""
"""Returns the virtual map from a windows kernel module."""
result = {}
system_va_type = module.get_enumeration('_MI_SYSTEM_VA_TYPE')
large_page_size = (module.context.layers[module.layer_name].page_size ** 2) // module.get_type("_MMPTE").size
+2 -1
View File
@@ -34,7 +34,8 @@ class YaraScanner(interfaces.layers.ScannerInterface):
class YaraScan(plugins.PluginInterface):
"""Runs all relevant plugins that provide time related information and orders the results by time"""
"""Runs all relevant plugins that provide time related information and
orders the results by time."""
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: