mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-13 05:07:38 +02:00
Consolidate identical code into a single symbol cache, add mac automagic list and pycharm reformat.
This commit is contained in:
Mike Auty
@@ -29,6 +29,11 @@ linux_automagic = ['ConstructionMagic',
|
||||
'LinuxSymbolCache',
|
||||
'LinuxSymbolFinder']
|
||||
|
||||
mac_automagic = ['ConstructionMagic',
|
||||
'LayerStacker',
|
||||
'MacSymbolCache',
|
||||
'MacSymbolFinder']
|
||||
|
||||
|
||||
def available(context: interfaces.context.ContextInterface) \
|
||||
-> typing.List[interfaces.automagic.AutomagicInterface]:
|
||||
@@ -59,6 +64,9 @@ def choose_automagic(automagics, plugin):
|
||||
elif plugin_category == 'linux':
|
||||
if amagic.__class__.__name__ in linux_automagic:
|
||||
output += [amagic]
|
||||
elif plugin_category == 'mac':
|
||||
if amagic.__class__.__name__ in mac_automagic:
|
||||
output += [amagic]
|
||||
else:
|
||||
return automagics
|
||||
vollog.info("Restricting automagics to: {}".format([x.__class__.__name__ for x in output]))
|
||||
|
||||
@@ -4,7 +4,7 @@ import typing
|
||||
import volatility.framework.objects.utility
|
||||
from volatility.framework import interfaces, constants, validity, exceptions, layers
|
||||
from volatility.framework import symbols, objects
|
||||
from volatility.framework.automagic import linux_symbol_cache
|
||||
from volatility.framework.automagic import symbol_cache
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.layers import intel, scanners
|
||||
from volatility.framework.symbols import linux
|
||||
@@ -21,13 +21,13 @@ class LinuxSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
config_path: str) -> None:
|
||||
super().__init__(context, config_path)
|
||||
self._requirements = [] # type: typing.List[typing.Tuple[str, interfaces.configuration.ConstructableRequirementInterface]]
|
||||
self._linux_banners_ = {} # type: linux_symbol_cache.LinuxBanners
|
||||
self._linux_banners_ = {} # type: symbol_cache.BannersType
|
||||
|
||||
@property
|
||||
def _linux_banners(self) -> linux_symbol_cache.LinuxBanners:
|
||||
def _linux_banners(self) -> symbol_cache.BannersType:
|
||||
"""Creates a cached copy of the results, but only it's been requested"""
|
||||
if not self._linux_banners_:
|
||||
self._linux_banners_ = linux_symbol_cache.LinuxSymbolCache.load_linux_banners()
|
||||
self._linux_banners_ = symbol_cache.LinuxSymbolCache.load_banners()
|
||||
return self._linux_banners_
|
||||
|
||||
def __call__(self,
|
||||
@@ -124,7 +124,7 @@ class LintelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
if isinstance(layer, intel.Intel):
|
||||
return None
|
||||
|
||||
linux_banners = linux_symbol_cache.LinuxSymbolCache.load_linux_banners()
|
||||
linux_banners = symbol_cache.LinuxSymbolCache.load_banners()
|
||||
mss = scanners.MultiStringScanner([x for x in linux_banners if x is not None])
|
||||
for _, banner in layer.scan(context = context, scanner = mss, progress_callback = progress_callback):
|
||||
dtb = None
|
||||
|
||||
@@ -1,17 +1,17 @@
|
||||
import sys, struct
|
||||
import logging
|
||||
import struct
|
||||
import typing
|
||||
|
||||
import volatility.framework.objects.utility
|
||||
from volatility.framework import layers, interfaces, constants, validity, exceptions
|
||||
from volatility.framework import symbols, objects
|
||||
from volatility.framework.automagic import mac_symbol_cache
|
||||
from volatility.framework import interfaces, constants, validity
|
||||
from volatility.framework import symbols
|
||||
from volatility.framework.automagic import symbol_cache
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.layers import intel, scanners
|
||||
from volatility.framework.symbols import mac
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MacSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
"""Mac symbol loader based on uname signature strings"""
|
||||
priority = 40
|
||||
@@ -24,11 +24,11 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
self._mac_banners_ = {} # type: mac_symbol_cache.MacBanners
|
||||
|
||||
@property
|
||||
def _mac_banners(self) -> mac_symbol_cache.MacBanners:
|
||||
def _mac_banners(self) -> symbol_cache.BannersType:
|
||||
"""Creates a cached copy of the results, but only it's been requested"""
|
||||
if not self._mac_banners_:
|
||||
self._mac_banners_ = mac_symbol_cache.MacSymbolCache.load_mac_banners()
|
||||
|
||||
self._mac_banners_ = symbol_cache.MacSymbolCache.load_mac_banners()
|
||||
|
||||
return self._mac_banners_
|
||||
|
||||
def __call__(self,
|
||||
@@ -37,7 +37,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
requirement: interfaces.configuration.RequirementInterface,
|
||||
progress_callback: validity.ProgressCallback = None) -> None:
|
||||
"""Searches for MacSymbolRequirements and attempt to populate them"""
|
||||
|
||||
|
||||
self._requirements = self.find_requirements(context, config_path, requirement,
|
||||
(requirements.TranslationLayerRequirement,
|
||||
requirements.SymbolRequirement),
|
||||
@@ -66,7 +66,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
progress_callback: validity.ProgressCallback = None) -> None:
|
||||
"""Accepts a context, config_path and SymbolRequirement, with a constructed layer_name
|
||||
and scans the layer for mac banners"""
|
||||
|
||||
|
||||
# Bomb out early if there's no banners
|
||||
if not self._mac_banners:
|
||||
return
|
||||
@@ -83,7 +83,7 @@ class MacSymbolFinder(interfaces.automagic.AutomagicInterface):
|
||||
# TODO: Fix this so it works for layers other than just Intel
|
||||
layer = context.memory[layer.config['memory_layer']]
|
||||
banner_list = layer.scan(context = context, scanner = mss, progress_callback = progress_callback)
|
||||
|
||||
|
||||
for _, banner in banner_list:
|
||||
vollog.debug("Identified banner: {}".format(repr(banner)))
|
||||
symbol_files = self._mac_banners.get(banner, None)
|
||||
@@ -126,10 +126,11 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
if isinstance(layer, intel.Intel):
|
||||
return None
|
||||
|
||||
mac_banners = mac_symbol_cache.MacSymbolCache.load_mac_banners()
|
||||
mac_banners = symbol_cache.MacSymbolCache.load_mac_banners()
|
||||
mss = scanners.MultiStringScanner([x for x in mac_banners if x is not None])
|
||||
|
||||
for banner_offset, banner in layer.scan(context = context, scanner = mss, progress_callback = progress_callback):
|
||||
|
||||
for banner_offset, banner in layer.scan(context = context, scanner = mss,
|
||||
progress_callback = progress_callback):
|
||||
dtb = None
|
||||
vollog.debug("Identified banner: {}".format(repr(banner)))
|
||||
|
||||
@@ -138,15 +139,15 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
isf_path = symbol_files[0]
|
||||
table_name = context.symbol_space.free_table_name('MacintelStacker')
|
||||
table = mac.MacKernelIntermedSymbols(context, 'temporary.' + table_name, name = table_name,
|
||||
isf_url = isf_path)
|
||||
isf_url = isf_path)
|
||||
context.symbol_space.append(table)
|
||||
kaslr_shift = MacUtilities.find_aslr(context, table_name, layer_name,
|
||||
banner, banner_offset, progress_callback = progress_callback)
|
||||
banner, banner_offset, progress_callback = progress_callback)
|
||||
|
||||
######################
|
||||
# ikelos: The following is what I tried to get the dtb, but couldn't figure out how to do
|
||||
# as you will see after the commented block of code is just a hardcoding of the DTB to my test sample's value
|
||||
#######################
|
||||
#######################
|
||||
'''
|
||||
bootpml4_addr = table.get_symbol("BootPML4").address + kaslr_shift
|
||||
|
||||
@@ -168,7 +169,7 @@ class MacintelStacker(interfaces.automagic.StackerLayerInterface):
|
||||
print("new dtb / idlepml4_addr = {:x".format(idlepml4_addr))
|
||||
sys.exit(1)
|
||||
'''
|
||||
|
||||
|
||||
dtb = 0x1ef6e000
|
||||
|
||||
# Build the new layer
|
||||
@@ -206,10 +207,10 @@ class MacUtilities(object):
|
||||
|
||||
def _scan_generator(self, context, layer_name, progress_callback):
|
||||
darwin_signature = b"Darwin Kernel Version \d{1,3}\.\d{1,3}\.\d{1,3}: [^\x00]+\x00"
|
||||
|
||||
|
||||
for offset in context.memory[layer_name].scan(scanner = scanners.RegExScanner(darwin_signature),
|
||||
context = context, progress_callback = progress_callback):
|
||||
|
||||
|
||||
banner = context.memory[layer_name].read(offset, 128)
|
||||
|
||||
idx = banner.find(b"\x00")
|
||||
@@ -228,17 +229,17 @@ class MacUtilities(object):
|
||||
progress_callback: validity.ProgressCallback = None) \
|
||||
-> typing.Tuple[int, int]:
|
||||
"""Determines the offset of the actual DTB in physical space and its symbol offset"""
|
||||
version_symbol = symbol_table + constants.BANG + 'version'
|
||||
version_symbol = symbol_table + constants.BANG + 'version'
|
||||
version_json_address = context.symbol_space.get_symbol(version_symbol).address
|
||||
version_phys_offset = MacUtilities.virtual_to_physical_address(version_json_address)
|
||||
|
||||
version_major_symbol = symbol_table + constants.BANG + 'version_major'
|
||||
version_major_json_address = context.symbol_space.get_symbol(version_major_symbol).address
|
||||
version_major_phys_offset = MacUtilities.virtual_to_physical_address(version_major_json_address)
|
||||
version_phys_offset = MacUtilities.virtual_to_physical_address(version_json_address)
|
||||
|
||||
version_minor_symbol = symbol_table + constants.BANG + 'version_minor'
|
||||
version_major_symbol = symbol_table + constants.BANG + 'version_major'
|
||||
version_major_json_address = context.symbol_space.get_symbol(version_major_symbol).address
|
||||
version_major_phys_offset = MacUtilities.virtual_to_physical_address(version_major_json_address)
|
||||
|
||||
version_minor_symbol = symbol_table + constants.BANG + 'version_minor'
|
||||
version_minor_json_address = context.symbol_space.get_symbol(version_minor_symbol).address
|
||||
version_minor_phys_offset = MacUtilities.virtual_to_physical_address(version_minor_json_address)
|
||||
version_minor_phys_offset = MacUtilities.virtual_to_physical_address(version_minor_json_address)
|
||||
|
||||
module = context.module(symbol_table, layer_name, 0)
|
||||
|
||||
@@ -264,20 +265,19 @@ class MacUtilities(object):
|
||||
minor = struct.unpack("<I", minor_string)[0]
|
||||
|
||||
if minor != banner_minor:
|
||||
conitnue
|
||||
continue
|
||||
|
||||
if aslr_shift & 0xfff != 0:
|
||||
continue
|
||||
|
||||
|
||||
aslr_shift = tmp_aslr_shift & 0xffffffff
|
||||
break
|
||||
|
||||
vollog.debug("Mac ASLR shift value determined: {:0x}".format(aslr_shift))
|
||||
|
||||
vollog.debug("Mac ASLR shift value determined: {:0x}".format(aslr_shift))
|
||||
|
||||
return aslr_shift
|
||||
|
||||
@classmethod
|
||||
def virtual_to_physical_address(cls, addr: int) -> int:
|
||||
"""Converts a virtual mac address to a physical one (does not account of ASLR)"""
|
||||
return addr - 0xffffff8000000000
|
||||
|
||||
|
||||
@@ -1,99 +0,0 @@
|
||||
import logging
|
||||
import os
|
||||
import pickle
|
||||
import typing
|
||||
import urllib
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from volatility.framework import constants, exceptions, interfaces
|
||||
from volatility.framework.symbols import intermed
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
MacBanners = typing.Dict[bytes, typing.List[str]]
|
||||
|
||||
class MacSymbolCache(interfaces.automagic.AutomagicInterface):
|
||||
"""Runs through all Mac symbols tables and caches their banners"""
|
||||
|
||||
# Since this is necessary for ConstructionMagic, we set a lower priority
|
||||
# The user would run it eventually either way, but running it first means it can be used that run
|
||||
priority = 0
|
||||
|
||||
@classmethod
|
||||
def load_mac_banners(cls) -> MacBanners:
|
||||
mac_banners = {} # type: MacBanners
|
||||
if os.path.exists(constants.MAC_BANNERS_PATH):
|
||||
with open(constants.MAC_BANNERS_PATH, "rb") as f:
|
||||
# We use pickle over JSON because we're dealing with bytes objects
|
||||
mac_banners.update(pickle.load(f))
|
||||
|
||||
# Remove possibilities that can't exist locally.
|
||||
remove_banners = []
|
||||
for banner in mac_banners:
|
||||
for path in mac_banners[banner]:
|
||||
url = urllib.parse.urlparse(path)
|
||||
if url.scheme == 'file' and not os.path.exists(urllib.request.url2pathname(url.path)):
|
||||
vollog.log(constants.LOGLEVEL_V,
|
||||
"Removing cached path {} for banner {}: file does not exist".format(path, banner))
|
||||
mac_banners[banner].remove(path)
|
||||
# This is probably excessive, but it's here if we need it
|
||||
# if url.scheme == 'jar':
|
||||
# zip_file, zip_path = url.path.split("!")
|
||||
# zip_file = urllib.parse.urlparse(zip_file).path
|
||||
# if ((not os.path.exists(zip_file)) or (zip_path not in zipfile.ZipFile(zip_file).namelist())):
|
||||
# vollog.log(constants.LOGLEVEL_V,
|
||||
# "Removing cached path {} for banner {}: file does not exist".format(path, banner))
|
||||
# mac_banners[banner].remove(path)
|
||||
|
||||
if not mac_banners[banner]:
|
||||
remove_banners.append(banner)
|
||||
for remove_banner in remove_banners:
|
||||
del mac_banners[remove_banner]
|
||||
return mac_banners
|
||||
|
||||
@classmethod
|
||||
def save_mac_banners(cls, mac_banners):
|
||||
|
||||
with open(constants.MAC_BANNERS_PATH, "wb") as f:
|
||||
pickle.dump(mac_banners, f)
|
||||
|
||||
def __call__(self, context, config_path, configurable, progress_callback = None):
|
||||
"""Runs the automagic over the configurable"""
|
||||
# We only need to be called once, so no recursion necessary
|
||||
macbanners = self.load_mac_banners()
|
||||
|
||||
cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url("mac"))
|
||||
vollog.info("Building mac cacheables...".format(cacheables))
|
||||
|
||||
for banner in macbanners:
|
||||
for json_file in macbanners[banner]:
|
||||
if json_file in cacheables:
|
||||
cacheables.remove(json_file)
|
||||
|
||||
total = len(cacheables)
|
||||
if total > 0:
|
||||
vollog.info("Building mac caches...")
|
||||
for current in range(total):
|
||||
#progress_callback(current * 100 / total, "Building mac caches")
|
||||
isf_url = cacheables[current]
|
||||
|
||||
try:
|
||||
# Loading the symbol table will be very slow until it's been validated
|
||||
isf = intermed.IntermediateSymbolTable(context, config_path, "temp", isf_url, validate = False)
|
||||
|
||||
# We should store the banner against the filename
|
||||
# We don't bother with the hash (it'll likely take too long to validate)
|
||||
# but we should check at least that the banner matches on load.
|
||||
banner = isf.get_symbol("version").constant_data
|
||||
vollog.log(constants.LOGLEVEL_V, "Caching banner {} for file {}".format(banner, isf_url))
|
||||
bannerlist = macbanners.get(banner, [])
|
||||
bannerlist.append(isf_url)
|
||||
macbanners[banner] = bannerlist
|
||||
except exceptions.SymbolError:
|
||||
pass
|
||||
|
||||
vollog.debug("writing mac banners: {}".format(macbanners))
|
||||
|
||||
# Rewrite the cached macbanners each run, since writing is faster than the cache validation portion
|
||||
self.save_mac_banners(macbanners)
|
||||
+47
-29
@@ -11,33 +11,36 @@ from volatility.framework.symbols import intermed
|
||||
|
||||
vollog = logging.getLogger(__name__)
|
||||
|
||||
LinuxBanners = typing.Dict[bytes, typing.List[str]]
|
||||
BannersType = typing.Dict[bytes, typing.List[str]]
|
||||
|
||||
|
||||
class LinuxSymbolCache(interfaces.automagic.AutomagicInterface):
|
||||
"""Runs through all Linux symbols tables and caches their banners"""
|
||||
class SymbolCache(interfaces.automagic.AutomagicInterface):
|
||||
"""Runs through all symbols tables and caches their banners"""
|
||||
|
||||
# Since this is necessary for ConstructionMagic, we set a lower priority
|
||||
# The user would run it eventually either way, but running it first means it can be used that run
|
||||
priority = 0
|
||||
os = None
|
||||
symbol_name = "linux_banner"
|
||||
banner_path = constants.LINUX_BANNERS_PATH
|
||||
|
||||
@classmethod
|
||||
def load_linux_banners(cls) -> LinuxBanners:
|
||||
linux_banners = {} # type: LinuxBanners
|
||||
if os.path.exists(constants.LINUX_BANNERS_PATH):
|
||||
with open(constants.LINUX_BANNERS_PATH, "rb") as f:
|
||||
def load_banners(cls) -> BannersType:
|
||||
banners = {} # type: BannersType
|
||||
if os.path.exists(cls.banner_path):
|
||||
with open(cls.banner_path, "rb") as f:
|
||||
# We use pickle over JSON because we're dealing with bytes objects
|
||||
linux_banners.update(pickle.load(f))
|
||||
banners.update(pickle.load(f))
|
||||
|
||||
# Remove possibilities that can't exist locally.
|
||||
remove_banners = []
|
||||
for banner in linux_banners:
|
||||
for path in linux_banners[banner]:
|
||||
for banner in banners:
|
||||
for path in banners[banner]:
|
||||
url = urllib.parse.urlparse(path)
|
||||
if url.scheme == 'file' and not os.path.exists(urllib.request.url2pathname(url.path)):
|
||||
vollog.log(constants.LOGLEVEL_V,
|
||||
"Removing cached path {} for banner {}: file does not exist".format(path, banner))
|
||||
linux_banners[banner].remove(path)
|
||||
banners[banner].remove(path)
|
||||
# This is probably excessive, but it's here if we need it
|
||||
# if url.scheme == 'jar':
|
||||
# zip_file, zip_path = url.path.split("!")
|
||||
@@ -45,37 +48,40 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface):
|
||||
# if ((not os.path.exists(zip_file)) or (zip_path not in zipfile.ZipFile(zip_file).namelist())):
|
||||
# vollog.log(constants.LOGLEVEL_V,
|
||||
# "Removing cached path {} for banner {}: file does not exist".format(path, banner))
|
||||
# linux_banners[banner].remove(path)
|
||||
# banners[banner].remove(path)
|
||||
|
||||
if not linux_banners[banner]:
|
||||
if not banners[banner]:
|
||||
remove_banners.append(banner)
|
||||
for remove_banner in remove_banners:
|
||||
del linux_banners[remove_banner]
|
||||
return linux_banners
|
||||
del banners[remove_banner]
|
||||
return banners
|
||||
|
||||
@classmethod
|
||||
def save_linux_banners(cls, linux_banners):
|
||||
def save_banners(cls, banners):
|
||||
|
||||
with open(constants.LINUX_BANNERS_PATH, "wb") as f:
|
||||
pickle.dump(linux_banners, f)
|
||||
with open(cls.banner_path, "wb") as f:
|
||||
pickle.dump(banners, f)
|
||||
|
||||
def __call__(self, context, config_path, configurable, progress_callback = None):
|
||||
"""Runs the automagic over the configurable"""
|
||||
if self.os is None:
|
||||
return
|
||||
|
||||
# We only need to be called once, so no recursion necessary
|
||||
linuxbanners = self.load_linux_banners()
|
||||
banners = self.load_banners()
|
||||
|
||||
cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url("linux"))
|
||||
cacheables = list(intermed.IntermediateSymbolTable.file_symbol_url(self.os))
|
||||
|
||||
for banner in linuxbanners:
|
||||
for json_file in linuxbanners[banner]:
|
||||
for banner in banners:
|
||||
for json_file in banners[banner]:
|
||||
if json_file in cacheables:
|
||||
cacheables.remove(json_file)
|
||||
|
||||
total = len(cacheables)
|
||||
if total > 0:
|
||||
vollog.info("Building linux caches...")
|
||||
vollog.info("Building {} caches...".format(self.os))
|
||||
for current in range(total):
|
||||
progress_callback(current * 100 / total, "Building linux caches")
|
||||
progress_callback(current * 100 / total, "Building {} caches".format(self.os))
|
||||
isf_url = cacheables[current]
|
||||
|
||||
try:
|
||||
@@ -85,13 +91,25 @@ class LinuxSymbolCache(interfaces.automagic.AutomagicInterface):
|
||||
# We should store the banner against the filename
|
||||
# We don't bother with the hash (it'll likely take too long to validate)
|
||||
# but we should check at least that the banner matches on load.
|
||||
banner = isf.get_symbol("linux_banner").constant_data
|
||||
banner = isf.get_symbol(self.symbol_name).constant_data
|
||||
vollog.log(constants.LOGLEVEL_V, "Caching banner {} for file {}".format(banner, isf_url))
|
||||
bannerlist = linuxbanners.get(banner, [])
|
||||
bannerlist = banners.get(banner, [])
|
||||
bannerlist.append(isf_url)
|
||||
linuxbanners[banner] = bannerlist
|
||||
banners[banner] = bannerlist
|
||||
except exceptions.SymbolError:
|
||||
pass
|
||||
|
||||
# Rewrite the cached linuxbanners each run, since writing is faster than the cache validation portion
|
||||
self.save_linux_banners(linuxbanners)
|
||||
# Rewrite the cached banners each run, since writing is faster than the cache validation portion
|
||||
self.save_banners(banners)
|
||||
|
||||
|
||||
class LinuxSymbolCache(SymbolCache):
|
||||
os = "linux"
|
||||
symbol_name = "linux_banner"
|
||||
banner_path = constants.LINUX_BANNERS_PATH
|
||||
|
||||
|
||||
class MacSymbolCache(SymbolCache):
|
||||
os = "mac"
|
||||
symbol_name = "version"
|
||||
banner_path = constants.MAC_BANNERS_PATH
|
||||
Reference in New Issue
Block a user