Automagic: Refactor ASLR finding for all symbol_finder using OSes

This commit is contained in:
Mike Auty
2020-06-10 19:39:20 +01:00
committed by ikelos
parent 863b9915c9
commit a6cd344e22
9 changed files with 72 additions and 101 deletions
+19 -35
View File
@@ -5,8 +5,8 @@
import logging
from typing import List, Optional, Tuple, Type
from volatility.framework import interfaces, constants, exceptions, layers
from volatility.framework import symbols, objects
from volatility.framework import interfaces, constants, exceptions
from volatility.framework import objects
from volatility.framework.automagic import symbol_cache, symbol_finder
from volatility.framework.layers import intel, scanners
from volatility.framework.symbols import linux
@@ -14,22 +14,6 @@ from volatility.framework.symbols import linux
vollog = logging.getLogger(__name__)
class LinuxBannerCache(symbol_cache.SymbolBannerCache):
"""Caches the banners found in the Linux symbol files."""
os = "linux"
symbol_name = "linux_banner"
banner_path = constants.LINUX_BANNERS_PATH
class LinuxSymbolFinder(symbol_finder.SymbolFinder):
"""Linux symbol loader based on uname signature strings."""
banner_config_key = "kernel_banner"
banner_cache = LinuxBannerCache
symbol_class = "volatility.framework.symbols.linux.LinuxKernelIntermedSymbols"
class LintelStacker(interfaces.automagic.StackerLayerInterface):
stack_order = 45
@@ -256,23 +240,6 @@ class LinuxUtilities(object):
yield fd_num, filp, full_path
@classmethod
def aslr_mask_symbol_table(cls,
context: interfaces.context.ContextInterface,
symbol_table: str,
layer_name: str,
aslr_shift = 0) -> str:
sym_layer = context.layers[layer_name]
if aslr_shift == 0:
if not isinstance(sym_layer, layers.intel.Intel):
raise TypeError("Layer name {} is not an intel space")
aslr_layer = sym_layer.config['memory_layer']
_, aslr_shift = cls.find_aslr(context, symbol_table, aslr_layer)
return symbols.mask_symbol_table(context, symbol_table, sym_layer.address_mask, aslr_shift)
@classmethod
def find_aslr(cls,
context: interfaces.context.ContextInterface,
@@ -319,3 +286,20 @@ class LinuxUtilities(object):
if addr > 0xffffffff80000000:
return addr - 0xffffffff80000000
return addr - 0xc0000000
class LinuxBannerCache(symbol_cache.SymbolBannerCache):
"""Caches the banners found in the Linux symbol files."""
os = "linux"
symbol_name = "linux_banner"
banner_path = constants.LINUX_BANNERS_PATH
class LinuxSymbolFinder(symbol_finder.SymbolFinder):
"""Linux symbol loader based on uname signature strings."""
banner_config_key = "kernel_banner"
banner_cache = LinuxBannerCache
symbol_class = "volatility.framework.symbols.linux.LinuxKernelIntermedSymbols"
find_aslr = lambda *args: LinuxUtilities.find_aslr(*args)[1]
+16 -45
View File
@@ -16,43 +16,6 @@ from volatility.framework.symbols import mac
vollog = logging.getLogger(__name__)
class MacBannerCache(symbol_cache.SymbolBannerCache):
"""Caches the banners found in the Mac symbol files."""
os = "mac"
symbol_name = "version"
banner_path = constants.MAC_BANNERS_PATH
class MacSymbolFinder(symbol_finder.SymbolFinder):
"""Mac symbol loader based on uname signature strings."""
banner_config_key = 'kernel_banner'
banner_cache = MacBannerCache
symbol_class = "volatility.framework.symbols.mac.MacKernelIntermedSymbols"
def _banner_scan(self,
context: interfaces.context.ContextInterface,
config_path: str,
requirement: interfaces.configuration.ConstructableRequirementInterface,
layer_name: str,
progress_callback: constants.ProgressCallback = None) -> None:
result = super()._banner_scan(context, config_path, requirement, layer_name, progress_callback)
new_symbol_table_name = context.config[path_join(config_path, requirement.name)]
sym_layer = context.layers[layer_name]
if not isinstance(sym_layer, layers.intel.Intel):
raise TypeError("Layer name {} is not an intel space")
aslr_layer = sym_layer.config['memory_layer']
aslr_shift = MacUtilities.find_aslr(context, new_symbol_table_name, aslr_layer)
masked_symbol_table_name = MacUtilities.aslr_mask_symbol_table(context, new_symbol_table_name, layer_name,
aslr_shift)
context.config[path_join(config_path, requirement.name)] = masked_symbol_table_name
return result
class MacintelStacker(interfaces.automagic.StackerLayerInterface):
stack_order = 45
@@ -198,14 +161,6 @@ class MacUtilities(object):
return mod_name, symbol_name
@classmethod
def aslr_mask_symbol_table(cls, context: interfaces.context.ContextInterface, symbol_table: str, layer_name: str,
aslr_shift) -> str:
sym_layer = context.layers[layer_name]
return symbols.mask_symbol_table(context, symbol_table, sym_layer.address_mask, aslr_shift)
@classmethod
def _scan_generator(cls, context, layer_name, progress_callback):
darwin_signature = rb"Darwin Kernel Version \d{1,3}\.\d{1,3}\.\d{1,3}: [^\x00]+\x00"
@@ -368,3 +323,19 @@ class MacUtilities(object):
current = current.member(attr = next_member).tqe_next
except exceptions.InvalidAddressException:
break
class MacBannerCache(symbol_cache.SymbolBannerCache):
"""Caches the banners found in the Mac symbol files."""
os = "mac"
symbol_name = "version"
banner_path = constants.MAC_BANNERS_PATH
class MacSymbolFinder(symbol_finder.SymbolFinder):
"""Mac symbol loader based on uname signature strings."""
banner_config_key = 'kernel_banner'
banner_cache = MacBannerCache
find_aslr = MacUtilities.find_aslr
symbol_class = "volatility.framework.symbols.mac.MacKernelIntermedSymbols"
@@ -3,9 +3,9 @@
#
import logging
from typing import Any, Iterable, List, Tuple, Type, Optional
from typing import Any, Iterable, List, Tuple, Type, Optional, Callable
from volatility.framework import interfaces, constants
from volatility.framework import interfaces, constants, layers, symbols
from volatility.framework.automagic import symbol_cache
from volatility.framework.configuration import requirements
from volatility.framework.layers import scanners
@@ -20,6 +20,7 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
banner_config_key = "banner" # type: str
banner_cache = None # type: Optional[Type[symbol_cache.SymbolBannerCache]]
symbol_class = None # type: Optional[str]
find_aslr = None # type: Optional[Callable]
def __init__(self, context: interfaces.context.ContextInterface, config_path: str) -> None:
super().__init__(context, config_path)
@@ -108,6 +109,17 @@ class SymbolFinder(interfaces.automagic.AutomagicInterface):
context.config[path_join(config_path, requirement.name, "isf_url")] = isf_path
# Construct the appropriate symbol table
requirement.construct(context, config_path)
# Apply the ASLR masking
if self.find_aslr:
unmasked_symbol_table_name = context.config[path_join(config_path, requirement.name)]
if not isinstance(layer, layers.intel.Intel):
raise TypeError("Layer name {} is not an intel space")
aslr_shift = self.find_aslr(context, unmasked_symbol_table_name, layer.config['memory_layer'])
masked_symbol_table_name = symbols.mask_symbol_table(context, unmasked_symbol_table_name,
layer_name, aslr_shift)
context.config[path_join(config_path, requirement.name)] = masked_symbol_table_name
break
else:
if symbol_files:
@@ -62,9 +62,7 @@ class Check_afinfo(plugins.PluginInterface):
yield var_name, "show", var.seq_show
def _generator(self):
masked_vmlinux_symbols = linux.LinuxUtilities.aslr_mask_symbol_table(self.context, self.config['vmlinux'], self.config['primary'])
vmlinux = contexts.Module(self.context, masked_vmlinux_symbols, self.config['primary'], 0)
vmlinux = contexts.Module(self.context, self.config['vmlinux'], self.config['primary'], 0)
op_members = vmlinux.get_type('file_operations').members
seq_members = vmlinux.get_type('seq_operations').members
@@ -122,10 +122,7 @@ class Check_syscall(plugins.PluginInterface):
# TODO - add finding and parsing unistd.h once cached file enumeration is added
def _generator(self):
masked_symbol_table = linux.LinuxUtilities.aslr_mask_symbol_table(self.context, self.config['vmlinux'],
self.config['primary'])
vmlinux = contexts.Module(self.context, masked_symbol_table, self.config['primary'], 0)
vmlinux = contexts.Module(self.context, self.config['vmlinux'], self.config['primary'], 0)
ptr_sz = vmlinux.get_type("pointer").size
if ptr_sz == 4:
+1 -3
View File
@@ -45,9 +45,7 @@ class Lsmod(plugins.PluginInterface):
This function will throw a SymbolError exception if kernel module support is not enabled.
"""
masked_vmlinux_symbols = linux.LinuxUtilities.aslr_mask_symbol_table(context, vmlinux_symbols, layer_name)
vmlinux = contexts.Module(context, masked_vmlinux_symbols, layer_name, 0)
vmlinux = contexts.Module(context, vmlinux_symbols, layer_name, 0)
modules = vmlinux.object_from_symbol(symbol_name = "modules").cast("list_head")
+1 -4
View File
@@ -45,14 +45,11 @@ class Lsof(plugins.PluginInterface):
yield (0, (pid, name, fd_num, full_path))
def run(self):
masked_vmlinux_symbols = linux.LinuxUtilities.aslr_mask_symbol_table(self.context, self.config['vmlinux'],
self.config['primary'])
filter_func = pslist.PsList.create_pid_filter([self.config.get('pid', None)])
return renderers.TreeGrid([("PID", int), ("Process", str), ("FD", int), ("Path", str)],
self._generator(
pslist.PsList.list_tasks(self.context,
self.config['primary'],
masked_vmlinux_symbols,
self.config['vmlinux'],
filter_func = filter_func)))
+1 -3
View File
@@ -75,9 +75,7 @@ class PsList(interfaces.plugins.PluginInterface):
Yields:
Process objects
"""
masked_vmlinux_symbols = linux.LinuxUtilities.aslr_mask_symbol_table(context, vmlinux_symbols, layer_name)
vmlinux = contexts.Module(context, masked_vmlinux_symbols, layer_name, 0)
vmlinux = contexts.Module(context, vmlinux_symbols, layer_name, 0)
init_task = vmlinux.object_from_symbol(symbol_name = "init_task")
+18 -2
View File
@@ -248,10 +248,26 @@ class SymbolSpace(interfaces.symbols.SymbolSpaceInterface):
def mask_symbol_table(context: interfaces.context.ContextInterface,
symbol_table_name: str,
address_mask: int = 0,
layer_name: str = "",
table_aslr_shift: int = 0) -> str:
"""Alters a symbol table, such that all symbols returned have their address
masked by the address mask."""
masked by the address mask.
Args:
context: Context that containing the symbol table and layers to be acted upon
symbol_table_name: Symbol table to mask
layer_name: Layer whose address mask will mask all symbol offsets
table_aslr_shift: Offset to add to all symbol addresses for ASLR
Returns:
Identifier for the new table that has been created
"""
layer = context.layers.get(layer_name, None)
if layer is None:
address_mask = 0
else:
address_mask = layer.address_mask
original_table = context.symbol_space[symbol_table_name]
new_table_name = context.symbol_space.free_table_name(original_table.name + '_masked'.format())
new_table = original_table.clone(new_table_name)