WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-28 20:44:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactor and add support for testing rekall.

Browse Source
This commit is contained in:
Mike Auty
2019-08-05 01:20:24 +01:00
parent f0a13c902e
commit f8587a08f2
1 changed files with 150 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
development/compare-vol.py
+150 -71
View File
@@ -1,7 +1,9 @@
import argparse
import csv
import hashlib
import os
import re
import shutil
import subprocess
import sys
import time
@@ -13,9 +15,10 @@ from typing import Dict, List
class VolatilityImage:
filepath: str = ""
vol2_profile: str = ""
vol2_time_imageinfo: int = None
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
@@ -23,41 +26,69 @@ 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 VolatilityTester:
class VolatilityTest:
short_name = "true"
long_name = "True"
def __init__(self, images, plugins, output_dir, vol2_path: str = None, vol3_path: str = None):
self.images = images
self.plugins = plugins
self.output_directory = output_dir
self.volatility2_path = vol2_path or self.output_directory
self.volatility3_path = vol3_path or self.output_directory
self.csv_writer = None
print("[?] Vol2 path", self.volatility2_path)
print("[?] Vol3 path", self.volatility3_path)
print("")
def __init__(self, path: str, output_directory: str) -> None:
self.path = path
self.output_directory = output_directory
def run_tests(self):
with open("volatility-timings.csv", 'w') as csvfile:
self.csv_writer = csv.writer(csvfile)
self.csv_writer.writerow([
"Image Path", "Plugin Name", "Volatility 3", "Volatility 3 (pypy)", "Volatility 2", "Image Info",
"Volatility 2 Total"
])
for image in self.images:
for plugin in self.plugins:
self.run_test(plugin, image)
def result_titles(self) -> List[str]:
return [self.long_name]
def run_test(self, plugin: VolatilityPlugin, image: VolatilityImage):
image_hash = hashlib.md5(bytes(image.filepath, "latin-1")).hexdigest()
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("[*] Testing {} {} with image {}".format(self.short_name, plugin.name, image.filepath))
os.chdir(self.path)
cmd = self.plugin_cmd(plugin, image)
start_time = time.perf_counter()
completed = subprocess.run(cmd, cwd = self.path, capture_output = True)
end_time = time.perf_counter()
total_time = end_time - start_time
print(" Tested {} {} with image {}: {}".format(self.short_name, plugin.name, image.filepath, total_time))
with open(
os.path.join(self.output_directory, '{}_{}_{}_stdout'.format(self.short_name, plugin.name, image_hash)),
"wb") as f:
f.write(completed.stdout)
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("[*] Testing volatility2 imageinfo with image {}".format(image.filepath))
os.chdir(self.volatility2_path)
print("[*] Testing {} imageinfo with image {}".format(self.short_name, 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.volatility2_path, capture_output = True)
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(" Tested volatility2 imageinfo with image {}: {}".format(image.filepath, end_time - start_time))
@@ -65,23 +96,29 @@ class VolatilityTester:
f.write(vol2_completed.stdout)
image.vol2_profile = re.search(b"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1]
# Volatility 2 Test
print("[*] Testing volatility2 {} with image {}".format(plugin.name, image.filepath))
os.chdir(self.volatility2_path)
cmd = ["python2", "-u", "vol.py", "-f", image.filepath, "--profile", image.vol2_profile
] + plugin.vol2_plugin_parameters + image.vol2_plugin_parameters.get(plugin.name, [])
start_time = time.perf_counter()
vol2_completed = subprocess.run(cmd, cwd = self.volatility2_path, capture_output = True)
end_time = time.perf_counter()
vol2_time = end_time - start_time
print(" Tested volatility2 {} with image {}: {}".format(plugin.name, image.filepath, vol2_time))
with open(os.path.join(self.output_directory, 'vol2_{}_{}_stdout'.format(plugin.name, image_hash)), "wb") as f:
f.write(vol2_completed.stdout)
# Volatility 3 Test
print("[*] Testing volatility3 {} with image {}".format(plugin.name, image.filepath))
os.chdir(self.volatility3_path)
cmd = [
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",
@@ -89,18 +126,14 @@ class VolatilityTester:
"-f",
image.filepath,
] + plugin.vol3_plugin_parameters + image.vol3_plugin_parameters.get(plugin.name, [])
start_time = time.perf_counter()
vol3_completed = subprocess.run(cmd, cwd = self.volatility3_path, capture_output = True)
end_time = time.perf_counter()
vol3_time = end_time - start_time
print(" Tested volatility3 {} with image {}: {}".format(plugin.name, image.filepath, vol3_time))
with open(os.path.join(self.output_directory, 'vol3_{}_{}_stdout'.format(plugin.name, image_hash)), "wb") as f:
f.write(vol3_completed.stdout)
# Volatility 3 Pypy3 Test
print("[*] Testing volatility3 (pypy) {} with image {}".format(plugin.name, image.filepath))
os.chdir(self.volatility3_path)
cmd = [
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",
@@ -108,23 +141,51 @@ class VolatilityTester:
"-f",
image.filepath,
] + plugin.vol3_plugin_parameters + image.vol3_plugin_parameters.get(plugin.name, [])
start_time = time.perf_counter()
vol3_pypy_completed = subprocess.run(cmd, cwd = self.volatility3_path, capture_output = True)
end_time = time.perf_counter()
vol3_pypy_time = end_time - start_time
print(" Tested volatility3 (pypy) {} with image {}: {}".format(plugin.name, image.filepath, vol3_pypy_time))
with open(os.path.join(self.output_directory, 'vol3_pypy_{}_{}_stdout'.format(plugin.name, image_hash)),
"wb") as f:
f.write(vol3_pypy_completed.stdout)
self.csv_writer.writerow([
image.filepath, plugin.name, vol3_time, vol3_pypy_time, vol2_time, image.vol2_imageinfo_time,
vol2_time + image.vol2_imageinfo_time
])
class VolatilityTester:
def __init__(self, images, plugins, output_dir, 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
self.tests = [
Volatility3Test(vol3_path, output_dir),
Volatility3PyPyTest(vol3_path, output_dir),
Volatility2Test(vol2_path, output_dir),
RekallTest(rekall_path, output_dir)
]
self.csv_writer = None
print("[?] Vol2 path", vol2_path)
print("[?] Vol3 path", vol3_path)
print("[?] Rekall path", rekall_path)
print("")
def run_tests(self):
with open("volatility-timings.csv", 'w') as csvfile:
self.csv_writer = csv.writer(csvfile)
self.csv_writer.writerow(["Image Path", "Plugin Name"] + [test.result_titles() for test in self.tests])
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.filepath, plugin.name] + results)
if __name__ == '__main__':
vt = VolatilityTester([VolatilityImage(filepath = x) for x in sys.argv[1:]], [
plugins = [
VolatilityPlugin(
name = "pslist", vol2_plugin_parameters = ["pslist"], vol3_plugin_parameters = ["windows.pslist"]),
VolatilityPlugin(
@@ -138,9 +199,15 @@ if __name__ == '__main__':
VolatilityPlugin(
name = "modules", vol2_plugin_parameters = ["modules"], vol3_plugin_parameters = ["windows.modules"]),
VolatilityPlugin(
name = "hivelist", vol2_plugin_parameters = ["hivelist"], vol3_plugin_parameters = ["registry.hivelist"]),
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"]),
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"]),
VolatilityPlugin(
@@ -149,6 +216,18 @@ if __name__ == '__main__':
VolatilityPlugin(
name = "printkey",
vol2_plugin_parameters = ["printkey", "-K", "Classes"],
vol3_plugin_parameters = ["registry.printkey", "--key", "Classes"])
], os.getcwd(), os.path.join(os.getcwd(), 'volatility'), os.path.join(os.getcwd(), 'volatility3'))
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())
parser.add_argument("--vol3path", type = str, default = os.path.join(os.getcwd(), 'volatility3'))
parser.add_argument("--vol2path", type = str, default = os.path.join(os.getcwd(), 'volatility'))
parser.add_argument("--rekallpath", type = str, default = os.path.join(os.getcwd(), 'rekall'))
parser.add_argument('images', metavar = 'IMAGE', type = str, nargs = '+', help = 'The list of images to compare')
args, excess = parser.parse_args()
vt = VolatilityTester([VolatilityImage(filepath = x) for x in args.images], plugins, args.output_dir, args.vol2path,
args.vol3path, args.rekallpath)
vt.run_tests()