import argparse import csv import hashlib import os import re import shutil import subprocess import time from dataclasses import dataclass, field from typing import Dict, List @dataclass class VolatilityImage: filepath: str = "" vol2_profile: str = "" vol2_imageinfo_time: float = None vol2_plugin_parameters: Dict[str, List[str]] = field(default_factory = dict) vol3_plugin_parameters: Dict[str, List[str]] = field(default_factory = dict) rekall_plugin_parameters: Dict[str, List[str]] = field(default_factory = dict) @dataclass class VolatilityPlugin: name: str = "" vol2_plugin_parameters: List[str] = field(default_factory = list) vol3_plugin_parameters: List[str] = field(default_factory = list) rekall_plugin_parameters: List[str] = field(default_factory = list) class VolatilityTest: short_name = "true" long_name = "True" def __init__(self, path: str, output_directory: str) -> None: self.path = path self.output_directory = output_directory def result_titles(self) -> List[str]: return [self.long_name] def create_prerequisites(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash: str) -> None: pass def create_results(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash: str) -> List[float]: self.create_prerequisites(plugin, image, image_hash) # Volatility 2 Test print(f"[*] Testing {self.short_name} {plugin.name} with image {image.filepath}") os.chdir(self.path) cmd = self.plugin_cmd(plugin, image) start_time = time.perf_counter() try: completed = subprocess.run(cmd, cwd = self.path, capture_output = True, timeout = 420) except subprocess.TimeoutExpired as excp: completed = excp end_time = time.perf_counter() total_time = end_time - start_time print(f" Tested {self.short_name} {plugin.name} with image {image.filepath}: {total_time}") with open( os.path.join(self.output_directory, f'{self.short_name}_{plugin.name}_{image_hash}_stdout'), "wb") as f: f.write(completed.stdout) if completed.stderr: with open( os.path.join(self.output_directory, f'{self.short_name}_{plugin.name}_{image_hash}_stderr'), "wb") as f: f.write(completed.stderr) return [total_time] def plugin_cmd(self, plugin: VolatilityPlugin, image: VolatilityImage): return ["true"] class Volatility2Test(VolatilityTest): short_name = "vol2" long_name = "Volatility 2" def plugin_cmd(self, plugin: VolatilityPlugin, image: VolatilityImage): return ["python2", "-u", "vol.py", "-f", image.filepath, "--profile", image.vol2_profile ] + plugin.vol2_plugin_parameters + image.vol2_plugin_parameters.get(plugin.name, []) def result_titles(self): return [self.long_name, "Imageinfo", f"{self.long_name} + Imageinfo"] def create_results(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash) -> List[float]: result = super().create_results(plugin, image, image_hash) result += [image.vol2_imageinfo_time, result[0] + image.vol2_imageinfo_time] return result def create_prerequisites(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash): # Volatility 2 image info if not image.vol2_profile: print(f"[*] Testing {self.short_name} imageinfo with image {image.filepath}") os.chdir(self.path) cmd = ["python2", "-u", "vol.py", "-f", image.filepath, "imageinfo"] start_time = time.perf_counter() vol2_completed = subprocess.run(cmd, cwd = self.path, capture_output = True) end_time = time.perf_counter() image.vol2_imageinfo_time = end_time - start_time print(f" Tested volatility2 imageinfo with image {image.filepath}: {end_time - start_time}") with open(os.path.join(self.output_directory, f'vol2_imageinfo_{image_hash}_stdout'), "wb") as f: f.write(vol2_completed.stdout) image.vol2_profile = re.search(b"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1] class RekallTest(VolatilityTest): short_name = "rekall" long_name = "Rekall" def plugin_cmd(self, plugin: VolatilityPlugin, image: VolatilityImage) -> List[str]: if not plugin.rekall_plugin_parameters: plugin.rekall_plugin_parameters = plugin.vol2_plugin_parameters if not image.rekall_plugin_parameters: image.rekall_plugin_parameters = image.vol2_plugin_parameters return ["rekall", "-f", image.filepath] + plugin.rekall_plugin_parameters + image.rekall_plugin_parameters.get( plugin.name, []) def create_prerequisites(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash: str) -> None: shutil.rmtree('/home/mike/.rekall_cache/sessions') class Volatility3Test(VolatilityTest): short_name = "vol3" long_name = "Volatility 3" def plugin_cmd(self, plugin: VolatilityPlugin, image: VolatilityImage) -> List[str]: return [ "python", "-u", "vol.py", "-q", "-f", image.filepath, ] + plugin.vol3_plugin_parameters + image.vol3_plugin_parameters.get(plugin.name, []) class Volatility3PyPyTest(VolatilityTest): short_name = "pypy" long_name = "Volatility 3 (PyPy)" def plugin_cmd(self, plugin: VolatilityPlugin, image: VolatilityImage) -> List[str]: return [ "pypy3", "-u", "vol.py", "-q", "-f", image.filepath, ] + plugin.vol3_plugin_parameters + image.vol3_plugin_parameters.get(plugin.name, []) class VolatilityTester: def __init__(self, images: List[VolatilityImage], plugins: List[VolatilityPlugin], frameworks: List[str], output_dir: str, vol2_path: str = None, vol3_path: str = None, rekall_path = None): self.images = images self.plugins = plugins if not vol2_path: vol2_path = output_dir if not vol3_path: vol3_path = output_dir if not rekall_path: rekall_path = output_dir available_tests = [ Volatility3Test(vol3_path, output_dir), Volatility3PyPyTest(vol3_path, output_dir), Volatility2Test(vol2_path, output_dir), RekallTest(rekall_path, output_dir) ] self.tests = [x for x in available_tests if x.short_name.lower() in frameworks] self.csv_writer = None print(f"[?] Vol2 path {vol2_path}") print(f"[?] Vol3 path {vol3_path}") print(f"[?] Rekall path {rekall_path}") print("") print(f"[?] Frameworks: {[x.long_name for x in self.tests]}") def run_tests(self): with open("volatility-timings.csv", 'w') as csvfile: self.csv_writer = csv.writer(csvfile) titles = ["Image Hash", "Image Path", "Plugin Name"] for test in self.tests: titles += test.result_titles() self.csv_writer.writerow(titles) for image in self.images: for plugin in self.plugins: self.run_test(plugin, image) def run_test(self, plugin: VolatilityPlugin, image: VolatilityImage): image_hash = hashlib.md5(bytes(image.filepath, "latin-1")).hexdigest() results = [] for test in self.tests: results += test.create_results(plugin, image, image_hash) self.csv_writer.writerow([image_hash, image.filepath, plugin.name] + results) if __name__ == '__main__': plugins = [ VolatilityPlugin(name = "pslist", vol2_plugin_parameters = ["pslist"], vol3_plugin_parameters = ["windows.pslist"]), VolatilityPlugin(name = "psscan", vol2_plugin_parameters = ["psscan"], vol3_plugin_parameters = ["windows.psscan"], rekall_plugin_parameters = ["psscan", "--scan_kernel"]), VolatilityPlugin(name = "driverscan", vol2_plugin_parameters = ["driverscan"], vol3_plugin_parameters = ["windows.driverscan"], rekall_plugin_parameters = ["driverscan", "--scan_kernel"]), VolatilityPlugin(name = "handles", vol2_plugin_parameters = ["handles"], vol3_plugin_parameters = ["windows.handles"]), VolatilityPlugin(name = "modules", vol2_plugin_parameters = ["modules"], vol3_plugin_parameters = ["windows.modules"]), VolatilityPlugin(name = "hivelist", vol2_plugin_parameters = ["hivelist"], vol3_plugin_parameters = ["registry.hivelist"], rekall_plugin_parameters = ["hives"]), VolatilityPlugin(name = "vadinfo", vol2_plugin_parameters = ["vadinfo"], vol3_plugin_parameters = ["windows.vadinfo"], rekall_plugin_parameters = ["vad"]), VolatilityPlugin(name = "modscan", vol2_plugin_parameters = ["modscan"], vol3_plugin_parameters = ["windows.modscan"], rekall_plugin_parameters = ["modscan", "--scan_kernel"]), VolatilityPlugin(name = "svcscan", vol2_plugin_parameters = ["svcscan"], vol3_plugin_parameters = ["windows.svcscan"], rekall_plugin_parameters = ["svcscan"]), VolatilityPlugin(name = "ssdt", vol2_plugin_parameters = ["ssdt"], vol3_plugin_parameters = ["windows.ssdt"]), VolatilityPlugin(name = "printkey", vol2_plugin_parameters = ["printkey", "-K", "Classes"], vol3_plugin_parameters = ["registry.printkey", "--key", "Classes"], rekall_plugin_parameters = ["printkey", "--key", "Classes"]) ] parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type = str, default = os.getcwd(), help = "Directory to store all results") parser.add_argument("--vol3path", type = str, default = os.path.join(os.getcwd(), 'volatility3'), help = "Path ot the volatility 3 directory") parser.add_argument("--vol2path", type = str, default = os.path.join(os.getcwd(), 'volatility'), help = "Path to the volatility 2 directory") parser.add_argument("--rekallpath", type = str, default = os.path.join(os.getcwd(), 'rekall'), help = "Path to the rekall directory") parser.add_argument("--frameworks", nargs = "+", type = str, choices = [x.short_name.lower() for x in VolatilityTest.__subclasses__()], default = [x.short_name.lower() for x in VolatilityTest.__subclasses__()], help = "A comma separated list of frameworks to test") parser.add_argument('images', metavar = 'IMAGE', type = str, nargs = '+', help = 'The list of images to compare') args = parser.parse_args() vt = VolatilityTester([VolatilityImage(filepath = x) for x in args.images], plugins, [x.lower() for x in args.frameworks], args.output_dir, args.vol2path, args.vol3path, args.rekallpath) vt.run_tests()