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

format with black

Browse Source
This commit is contained in:
Arthur Deierlein
2024-12-17 22:55:39 +01:00
parent eee015033a
commit f6c852c478
8 changed files with 462 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
development/banner_server.py
+31 -14
View File
@@ -28,10 +28,10 @@ class BannerCacheGenerator:
def run(self):
context = contexts.Context()
json_output = {'version': 1}
json_output = {"version": 1}
path = self._path
filename = '*'
filename = "*"
for banner_cache in [linux.LinuxBannerCache, mac.MacBannerCache]:
sub_path = banner_cache.os
@@ -39,37 +39,54 @@ class BannerCacheGenerator:
for extension in constants.ISF_EXTENSIONS:
# Hopefully these will not be large lists, otherwise this might be slow
try:
for found in pathlib.Path(path).joinpath(sub_path).resolve().rglob(filename + extension):
for found in (
pathlib.Path(path)
.joinpath(sub_path)
.resolve()
.rglob(filename + extension)
):
potentials.append(found.as_uri())
except FileNotFoundError:
# If there's no linux symbols, don't cry about it
pass
new_banners = banner_cache.read_new_banners(context, 'BannerServer', potentials, banner_cache.symbol_name,
banner_cache.os, progress_callback = PrintedProgress())
new_banners = banner_cache.read_new_banners(
context,
"BannerServer",
potentials,
banner_cache.symbol_name,
banner_cache.os,
progress_callback=PrintedProgress(),
)
result_banners = {}
for new_banner in new_banners:
# Only accept file schemes
value = [self.convert_url(url) for url in new_banners[new_banner] if
urllib.parse.urlparse(url).scheme == 'file']
value = [
self.convert_url(url)
for url in new_banners[new_banner]
if urllib.parse.urlparse(url).scheme == "file"
]
if value and new_banner:
# Convert files into URLs
result_banners[str(base64.b64encode(new_banner), 'latin-1')] = value
result_banners[str(base64.b64encode(new_banner), "latin-1")] = value
json_output[banner_cache.os] = result_banners
output_path = os.path.join(self._path, 'banners.json')
with open(output_path, 'w') as fp:
output_path = os.path.join(self._path, "banners.json")
with open(output_path, "w") as fp:
vollog.warning(f"Banners file written to {output_path}")
json.dump(json_output, fp)
if __name__ == '__main__':
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('--path', default = os.path.dirname(__file__))
parser.add_argument('--urlprefix', help = 'Web prefix that will eventually serve the ISF files',
default = 'http://localhost/symbols')
parser.add_argument("--path", default=os.path.dirname(__file__))
parser.add_argument(
"--urlprefix",
help="Web prefix that will eventually serve the ISF files",
default="http://localhost/symbols",
)
args = parser.parse_args()
development/compare-vol.py
+224 -116
View File
@@ -15,17 +15,17 @@ 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)
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)
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:
@@ -39,32 +39,50 @@ class VolatilityTest:
def result_titles(self) -> List[str]:
return [self.long_name]
def create_prerequisites(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash: str) -> None:
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]:
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}")
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)
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}")
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:
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:
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]
@@ -77,31 +95,57 @@ class Volatility2Test(VolatilityTest):
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, [])
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]:
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):
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}")
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)
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:
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(rb"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1]
image.vol2_profile = re.search(
rb"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout
)[1]
class RekallTest(VolatilityTest):
@@ -113,11 +157,16 @@ class RekallTest(VolatilityTest):
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, [])
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')
def create_prerequisites(
self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash: str
) -> None:
shutil.rmtree("/home/mike/.rekall_cache/sessions")
class Volatility3Test(VolatilityTest):
@@ -125,14 +174,18 @@ class Volatility3Test(VolatilityTest):
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, [])
return (
[
"python",
"-u",
"vol.py",
"-q",
"-f",
image.filepath,
]
+ plugin.vol3_plugin_parameters
+ image.vol3_plugin_parameters.get(plugin.name, [])
)
class Volatility3PyPyTest(VolatilityTest):
@@ -140,26 +193,32 @@ class Volatility3PyPyTest(VolatilityTest):
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, [])
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):
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:
@@ -172,7 +231,7 @@ class VolatilityTester:
Volatility3Test(vol3_path, output_dir),
Volatility3PyPyTest(vol3_path, output_dir),
Volatility2Test(vol2_path, output_dir),
RekallTest(rekall_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
@@ -183,7 +242,7 @@ class VolatilityTester:
print(f"[?] Frameworks: {[x.long_name for x in self.tests]}")
def run_tests(self):
with open("volatility-timings.csv", 'w') as csvfile:
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:
@@ -203,72 +262,121 @@ class VolatilityTester:
self.csv_writer.writerow([image_hash, image.filepath, plugin.name] + results)
if __name__ == '__main__':
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"])
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')
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 = 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()
development/mac-kdk/parse_pbzx2.py
+26 -24
View File
@@ -12,7 +12,7 @@ import struct
import sys
def seekread(f, offset = None, length = 0, relative = True):
def seekread(f, offset=None, length=0, relative=True):
if offset is not None:
# offset provided, let's seek
f.seek(offset, [0, 1, 2][relative])
@@ -23,55 +23,57 @@ def seekread(f, offset = None, length = 0, relative = True):
def parse_pbzx(pbzx_path):
section = 0
xar_out_path = f'{pbzx_path}.part{section:02d}.cpio.xz'
with open(pbzx_path, 'rb') as f:
xar_out_path = f"{pbzx_path}.part{section:02d}.cpio.xz"
with open(pbzx_path, "rb") as f:
# pbzx = f.read()
# f.close()
magic = seekread(f, length = 4)
if magic != 'pbzx':
magic = seekread(f, length=4)
if magic != "pbzx":
raise RuntimeError("Error: Not a pbzx file")
# Read 8 bytes for initial flags
flags = seekread(f, length = 8)
flags = seekread(f, length=8)
# Interpret the flags as a 64-bit big-endian unsigned int
flags = struct.unpack('>Q', flags)[0]
flags = struct.unpack(">Q", flags)[0]
while flags & (1 << 24):
with open(xar_out_path, 'wb') as xar_f:
with open(xar_out_path, "wb") as xar_f:
xar_f.seek(0, os.SEEK_END)
# Read in more flags
flags = seekread(f, length = 8)
flags = struct.unpack('>Q', flags)[0]
flags = seekread(f, length=8)
flags = struct.unpack(">Q", flags)[0]
# Read in length
f_length = seekread(f, length = 8)
f_length = struct.unpack('>Q', f_length)[0]
xzmagic = seekread(f, length = 6)
if xzmagic != '\xfd7zXZ\x00':
f_length = seekread(f, length=8)
f_length = struct.unpack(">Q", f_length)[0]
xzmagic = seekread(f, length=6)
if xzmagic != "\xfd7zXZ\x00":
# This isn't xz content, this is actually _raw decompressed cpio_ chunk of 16MB in size...
# Let's back up ...
seekread(f, offset = -6, length = 0)
seekread(f, offset=-6, length=0)
# ... and split it out ...
f_content = seekread(f, length = f_length)
f_content = seekread(f, length=f_length)
section += 1
decomp_out = f'{pbzx_path}.part{section:02d}.cpio'
with open(decomp_out, 'wb') as g:
decomp_out = f"{pbzx_path}.part{section:02d}.cpio"
with open(decomp_out, "wb") as g:
g.write(f_content)
# Now to start the next section, which should hopefully be .xz (we'll just assume it is ...)
section += 1
xar_out_path = f'{pbzx_path}.part{section:02d}.cpio.xz'
xar_out_path = f"{pbzx_path}.part{section:02d}.cpio.xz"
else:
f_length -= 6
# This part needs buffering
f_content = seekread(f, length = f_length)
tail = seekread(f, offset = -2, length = 2)
f_content = seekread(f, length=f_length)
tail = seekread(f, offset=-2, length=2)
xar_f.write(xzmagic)
xar_f.write(f_content)
if tail != 'YZ':
if tail != "YZ":
raise RuntimeError("Error: Footer is not xar file footer")
def main():
parse_pbzx(sys.argv[1])
print("Now xz decompress the .xz chunks, then 'cat' them all together in order into a single new.cpio file")
print(
"Now xz decompress the .xz chunks, then 'cat' them all together in order into a single new.cpio file"
)
if __name__ == '__main__':
if __name__ == "__main__":
main()
development/pdbparse-to-json.py
+118 -70
View File
@@ -13,10 +13,10 @@ import pdbparse.undecorate
logger = logging.getLogger(__name__)
logger.setLevel(1)
if __name__ == '__main__':
if __name__ == "__main__":
console = logging.StreamHandler()
console.setLevel(1)
formatter = logging.Formatter('%(levelname)-8s %(name)-12s: %(message)s')
formatter = logging.Formatter("%(levelname)-8s %(name)-12s: %(message)s")
console.setFormatter(formatter)
logger.addHandler(console)
@@ -25,12 +25,12 @@ class PDBRetreiver:
def retreive_pdb(self, guid: str, file_name: str) -> Optional[str]:
logger.info("Download PDB file...")
file_name = ".".join(file_name.split(".")[:-1] + ['pdb'])
for sym_url in ['http://msdl.microsoft.com/download/symbols']:
file_name = ".".join(file_name.split(".")[:-1] + ["pdb"])
for sym_url in ["http://msdl.microsoft.com/download/symbols"]:
url = sym_url + f"/{file_name}/{guid}/"
result = None
for suffix in [file_name[:-1] + '_', file_name]:
for suffix in [file_name[:-1] + "_", file_name]:
try:
logger.debug("Attempting to retrieve %s", url + suffix)
result, _ = request.urlretrieve(url + suffix)
@@ -69,7 +69,7 @@ class PDBConvertor:
"float": "float",
"double": "float",
"long double": "float",
"void": "void"
"void": "void",
}
base_type_size = {
@@ -122,13 +122,18 @@ class PDBConvertor:
self._seen_ctypes.add(ctype)
return self.ctype[ctype]
def lookup_ctype_pointers(self, ctype_pointer: str) -> Dict[str, Union[str, Dict[str, str]]]:
base_type = ctype_pointer.replace('32P', '').replace('64P', '')
def lookup_ctype_pointers(
self, ctype_pointer: str
) -> Dict[str, Union[str, Dict[str, str]]]:
base_type = ctype_pointer.replace("32P", "").replace("64P", "")
if base_type == ctype_pointer:
# We raise a KeyError, because we've been asked about a type that isn't a pointer
raise KeyError
self._seen_ctypes.add(base_type)
return {"kind": "pointer", "subtype": {"kind": "base", "name": self.ctype[base_type]}}
return {
"kind": "pointer",
"subtype": {"kind": "base", "name": self.ctype[base_type]},
}
def read_pdb(self) -> Dict:
"""Reads in the PDB file and forms essentially a python dictionary of necessary data"""
@@ -137,31 +142,31 @@ class PDBConvertor:
"enums": self.read_enums(),
"metadata": self.generate_metadata(),
"symbols": self.read_symbols(),
"base_types": self.read_basetypes()
"base_types": self.read_basetypes(),
}
return output
def generate_metadata(self) -> Dict[str, Any]:
"""Generates the metadata necessary for this object"""
dbg = self._pdb.STREAM_DBI
last_bytes = str(binascii.hexlify(self._pdb.STREAM_PDB.GUID.Data4), 'ascii')[-16:]
guidstr = f'{self._pdb.STREAM_PDB.GUID.Data1:08x}{self._pdb.STREAM_PDB.GUID.Data2:04x}{self._pdb.STREAM_PDB.GUID.Data3:04x}{last_bytes}'
last_bytes = str(binascii.hexlify(self._pdb.STREAM_PDB.GUID.Data4), "ascii")[
-16:
]
guidstr = f"{self._pdb.STREAM_PDB.GUID.Data1:08x}{self._pdb.STREAM_PDB.GUID.Data2:04x}{self._pdb.STREAM_PDB.GUID.Data3:04x}{last_bytes}"
pdb_data = {
"GUID": guidstr.upper(),
"age": self._pdb.STREAM_PDB.Age,
"database": "ntkrnlmp.pdb",
"machine_type": int(dbg.machine)
"machine_type": int(dbg.machine),
}
result = {
"format": "6.0.0",
"producer": {
"datetime": datetime.datetime.now().isoformat(),
"name": "pdbconv",
"version": "0.1.0"
"version": "0.1.0",
},
"windows": {
"pdb": pdb_data
}
"windows": {"pdb": pdb_data},
}
return result
@@ -172,16 +177,21 @@ class PDBConvertor:
stream = self._pdb.STREAM_TPI
for type_index in stream.types:
user_type = stream.types[type_index]
if (user_type.leaf_type == "LF_ENUM" and not user_type.prop.fwdref):
if user_type.leaf_type == "LF_ENUM" and not user_type.prop.fwdref:
output.update(self._format_enum(user_type))
return output
def _format_enum(self, user_enum):
output = {
user_enum.name: {
'base': self.lookup_ctype(user_enum.utype),
'size': self._determine_size(user_enum.utype),
'constants': dict([(enum.name, enum.enum_value) for enum in user_enum.fieldlist.substructs])
"base": self.lookup_ctype(user_enum.utype),
"size": self._determine_size(user_enum.utype),
"constants": dict(
[
(enum.name, enum.enum_value)
for enum in user_enum.fieldlist.substructs
]
),
}
}
return output
@@ -201,7 +211,7 @@ class PDBConvertor:
omap = None
for sym in self._pdb.STREAM_GSYM.globals:
if not hasattr(sym, 'offset'):
if not hasattr(sym, "offset"):
continue
try:
virt_base = sects[sym.segment - 1].VirtualAddress
@@ -222,9 +232,9 @@ class PDBConvertor:
stream = self._pdb.STREAM_TPI
for type_index in stream.types:
user_type = stream.types[type_index]
if (user_type.leaf_type == "LF_STRUCTURE" and not user_type.prop.fwdref):
if user_type.leaf_type == "LF_STRUCTURE" and not user_type.prop.fwdref:
output.update(self._format_usertype(user_type, "struct"))
elif (user_type.leaf_type == "LF_UNION" and not user_type.prop.fwdref):
elif user_type.leaf_type == "LF_UNION" and not user_type.prop.fwdref:
output.update(self._format_usertype(user_type, "union"))
return output
@@ -232,16 +242,22 @@ class PDBConvertor:
"""Produces a single usertype"""
fields: Dict[str, Dict[str, Any]] = {}
[fields.update(self._format_field(s)) for s in usertype.fieldlist.substructs]
return {usertype.name: {'fields': fields, 'kind': kind, 'size': usertype.size}}
return {usertype.name: {"fields": fields, "kind": kind, "size": usertype.size}}
def _format_field(self, field) -> Dict[str, Dict[str, Any]]:
return {field.name: {"offset": field.offset, "type": self._format_kind(field.index)}}
return {
field.name: {"offset": field.offset, "type": self._format_kind(field.index)}
}
def _determine_size(self, field):
output = None
if isinstance(field, str):
output = self.base_type_size[field]
elif (field.leaf_type == "LF_STRUCTURE" or field.leaf_type == "LF_ARRAY" or field.leaf_type == "LF_UNION"):
elif (
field.leaf_type == "LF_STRUCTURE"
or field.leaf_type == "LF_ARRAY"
or field.leaf_type == "LF_UNION"
):
output = field.size
elif field.leaf_type == "LF_POINTER":
output = self.base_type_size[field.ptr_attr.type]
@@ -255,6 +271,7 @@ class PDBConvertor:
output = self._determine_size(field.index)
if output is None:
import pdb
pdb.set_trace()
raise ValueError(f"Unknown size for field: {field.name}")
return output
@@ -266,36 +283,37 @@ class PDBConvertor:
output = self.lookup_ctype_pointers(kind)
except KeyError:
try:
output = {'kind': 'base', 'name': self.lookup_ctype(kind)}
output = {"kind": "base", "name": self.lookup_ctype(kind)}
except KeyError:
output = {'kind': 'base', 'name': kind}
elif kind.leaf_type == 'LF_MODIFIER':
output = {"kind": "base", "name": kind}
elif kind.leaf_type == "LF_MODIFIER":
output = self._format_kind(kind.modified_type)
elif kind.leaf_type == 'LF_STRUCTURE':
output = {'kind': 'struct', 'name': kind.name}
elif kind.leaf_type == 'LF_UNION':
output = {'kind': 'union', 'name': kind.name}
elif kind.leaf_type == 'LF_BITFIELD':
elif kind.leaf_type == "LF_STRUCTURE":
output = {"kind": "struct", "name": kind.name}
elif kind.leaf_type == "LF_UNION":
output = {"kind": "union", "name": kind.name}
elif kind.leaf_type == "LF_BITFIELD":
output = {
'kind': 'bitfield',
'type': self._format_kind(kind.base_type),
'bit_length': kind.length,
'bit_position': kind.position
"kind": "bitfield",
"type": self._format_kind(kind.base_type),
"bit_length": kind.length,
"bit_position": kind.position,
}
elif kind.leaf_type == 'LF_POINTER':
output = {'kind': 'pointer', 'subtype': self._format_kind(kind.utype)}
elif kind.leaf_type == 'LF_ARRAY':
elif kind.leaf_type == "LF_POINTER":
output = {"kind": "pointer", "subtype": self._format_kind(kind.utype)}
elif kind.leaf_type == "LF_ARRAY":
output = {
'kind': 'array',
'count': kind.size // self._determine_size(kind.element_type),
'subtype': self._format_kind(kind.element_type)
"kind": "array",
"count": kind.size // self._determine_size(kind.element_type),
"subtype": self._format_kind(kind.element_type),
}
elif kind.leaf_type == 'LF_ENUM':
output = {'kind': 'enum', 'name': kind.name}
elif kind.leaf_type == 'LF_PROCEDURE':
output = {'kind': "function"}
elif kind.leaf_type == "LF_ENUM":
output = {"kind": "enum", "name": kind.name}
elif kind.leaf_type == "LF_PROCEDURE":
output = {"kind": "function"}
else:
import pdb
pdb.set_trace()
return output
@@ -305,40 +323,70 @@ class PDBConvertor:
if "64" in self._pdb.STREAM_DBI.machine:
ptr_size = 8
output = {"pointer": {"endian": "little", "kind": "int", "signed": False, "size": ptr_size}}
output = {
"pointer": {
"endian": "little",
"kind": "int",
"signed": False,
"size": ptr_size,
}
}
for index in self._seen_ctypes:
output[self.ctype[index]] = {
"endian": "little",
"kind": self.ctype_python_types.get(self.ctype[index], "int"),
"signed": False if "_U" in index else True,
"size": self.base_type_size[index]
"size": self.base_type_size[index],
}
return output
if __name__ == '__main__':
parser = argparse.ArgumentParser(description = "Convertor for PDB files to Volatility 3 Intermediate Symbol Format")
parser.add_argument("-o", "--output", metavar = "OUTPUT", help = "Filename for data output", required = True)
file_group = parser.add_argument_group("file", description = "File-based conversion of PDB to ISF")
file_group.add_argument("-f", "--file", metavar = "FILE", help = "PDB file to translate to ISF")
data_group = parser.add_argument_group("data", description = "Convert based on a GUID and filename pattern")
data_group.add_argument("-p", "--pattern", metavar = "PATTERN", help = "Filename pattern to recover PDB file")
data_group.add_argument("-g",
"--guid",
metavar = "GUID",
help = "GUID + Age string for the required PDB file",
default = None)
data_group.add_argument("-k",
"--keep",
action = "store_true",
default = False,
help = "Keep the downloaded PDB file")
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Convertor for PDB files to Volatility 3 Intermediate Symbol Format"
)
parser.add_argument(
"-o",
"--output",
metavar="OUTPUT",
help="Filename for data output",
required=True,
)
file_group = parser.add_argument_group(
"file", description="File-based conversion of PDB to ISF"
)
file_group.add_argument(
"-f", "--file", metavar="FILE", help="PDB file to translate to ISF"
)
data_group = parser.add_argument_group(
"data", description="Convert based on a GUID and filename pattern"
)
data_group.add_argument(
"-p",
"--pattern",
metavar="PATTERN",
help="Filename pattern to recover PDB file",
)
data_group.add_argument(
"-g",
"--guid",
metavar="GUID",
help="GUID + Age string for the required PDB file",
default=None,
)
data_group.add_argument(
"-k",
"--keep",
action="store_true",
default=False,
help="Keep the downloaded PDB file",
)
args = parser.parse_args()
delfile = False
filename = None
if args.guid is not None and args.pattern is not None:
filename = PDBRetreiver().retreive_pdb(guid = args.guid, file_name = args.pattern)
filename = PDBRetreiver().retreive_pdb(guid=args.guid, file_name=args.pattern)
delfile = True
elif args.file:
filename = args.file
@@ -351,7 +399,7 @@ if __name__ == '__main__':
convertor = PDBConvertor(filename)
with open(args.output, "w") as f:
json.dump(convertor.read_pdb(), f, indent = 2, sort_keys = True)
json.dump(convertor.read_pdb(), f, indent=2, sort_keys=True)
if args.keep:
print(f"Temporary PDB file: {filename}")
development/schema_validate.py
+4 -4
View File
@@ -9,7 +9,7 @@ sys.path += ".."
console = logging.StreamHandler()
console.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(levelname)-8s %(name)-12s: %(message)s')
formatter = logging.Formatter("%(levelname)-8s %(name)-12s: %(message)s")
console.setFormatter(formatter)
logger = logging.getLogger("")
@@ -18,10 +18,10 @@ logger.setLevel(logging.DEBUG)
from volatility3 import schemas # noqa: E402
if __name__ == '__main__':
if __name__ == "__main__":
parser = argparse.ArgumentParser("Validates ")
parser.add_argument("-s", "--schema", dest = "schema", default = None)
parser.add_argument("filenames", metavar = "FILE", nargs = '+')
parser.add_argument("-s", "--schema", dest="schema", default=None)
parser.add_argument("filenames", metavar="FILE", nargs="+")
args = parser.parse_args()
development/stock-linux-json.py
+56 -42
View File
@@ -9,7 +9,7 @@ import requests
import rpmfile
from debian import debfile
DWARF2JSON = './dwarf2json'
DWARF2JSON = "./dwarf2json"
class Downloader:
@@ -17,7 +17,7 @@ class Downloader:
def __init__(self, url_lists: List[List[str]]) -> None:
self.url_lists = url_lists
def download_lists(self, keep = False):
def download_lists(self, keep=False):
for url_list in self.url_lists:
print("Downloading files...")
files_for_processing = self.download_list(url_list)
@@ -35,43 +35,45 @@ class Downloader:
with tempfile.NamedTemporaryFile() as archivedata:
archivedata.write(data.content)
archivedata.seek(0)
if url.endswith('.rpm'):
if url.endswith(".rpm"):
processed_files[url] = self.process_rpm(archivedata)
elif url.endswith('.deb'):
elif url.endswith(".deb"):
processed_files[url] = self.process_deb(archivedata)
return processed_files
def process_rpm(self, archivedata) -> Optional[str]:
rpm = rpmfile.RPMFile(fileobj = archivedata)
rpm = rpmfile.RPMFile(fileobj=archivedata)
member = None
extracted = None
for member in rpm.getmembers():
if 'vmlinux' in member.name or 'System.map' in member.name:
if "vmlinux" in member.name or "System.map" in member.name:
print(f" - Extracting {member.name}")
extracted = rpm.extractfile(member)
break
if not member or not extracted:
return None
with tempfile.NamedTemporaryFile(delete = False,
prefix = 'vmlinux' if 'vmlinux' in member.name else 'System.map') as output:
with tempfile.NamedTemporaryFile(
delete=False, prefix="vmlinux" if "vmlinux" in member.name else "System.map"
) as output:
print(f" - Writing to {output.name}")
output.write(extracted.read())
return output.name
def process_deb(self, archivedata) -> Optional[str]:
deb = debfile.DebFile(fileobj = archivedata)
deb = debfile.DebFile(fileobj=archivedata)
member = None
extracted = None
for member in deb.data.tgz().getmembers():
if member.name.endswith('vmlinux') or 'System.map' in member.name:
if member.name.endswith("vmlinux") or "System.map" in member.name:
print(f" - Extracting {member.name}")
extracted = deb.data.get_file(member.name)
break
if not member or not extracted:
return None
with tempfile.NamedTemporaryFile(delete = False,
prefix = 'vmlinux' if 'vmlinux' in member.name else 'System.map') as output:
with tempfile.NamedTemporaryFile(
delete=False, prefix="vmlinux" if "vmlinux" in member.name else "System.map"
) as output:
print(f" - Writing to {output.name}")
output.write(extracted.read())
return output.name
@@ -83,43 +85,55 @@ class Downloader:
if named_files[i] is None:
print(f"FAILURE: None encountered for {i}")
return
args = [DWARF2JSON, 'linux']
output_filename = 'unknown-kernel.json'
args = [DWARF2JSON, "linux"]
output_filename = "unknown-kernel.json"
for named_file in named_files:
prefix = '--system-map'
if 'System' not in named_files[named_file]:
prefix = '--elf'
output_filename = './' + '-'.join((named_file.split('/')[-1]).split('-')[2:])[:-4] + '.json.xz'
prefix = "--system-map"
if "System" not in named_files[named_file]:
prefix = "--elf"
output_filename = (
"./"
+ "-".join((named_file.split("/")[-1]).split("-")[2:])[:-4]
+ ".json.xz"
)
args += [prefix, named_files[named_file]]
print(f" - Running {args}")
proc = subprocess.run(args, capture_output = True)
proc = subprocess.run(args, capture_output=True)
print(f" - Writing to {output_filename}")
with lzma.open(output_filename, 'w') as f:
with lzma.open(output_filename, "w") as f:
f.write(proc.stdout)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description = "Takes a list of URLs for Centos and downloads them")
parser.add_argument("-f",
"--file",
dest = 'filename',
metavar = "FILENAME",
help = "Filename to be read",
required = True)
parser.add_argument("-d",
"--dwarf2json",
dest = 'dwarfpath',
metavar = "PATH",
default = DWARF2JSON,
help = "Path to the dwarf2json binary",
required = True)
parser.add_argument("-k",
"--keep",
dest = 'keep',
action = 'store_true',
help = 'Keep extracted temporary files after completion',
default = False)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Takes a list of URLs for Centos and downloads them"
)
parser.add_argument(
"-f",
"--file",
dest="filename",
metavar="FILENAME",
help="Filename to be read",
required=True,
)
parser.add_argument(
"-d",
"--dwarf2json",
dest="dwarfpath",
metavar="PATH",
default=DWARF2JSON,
help="Path to the dwarf2json binary",
required=True,
)
parser.add_argument(
"-k",
"--keep",
dest="keep",
action="store_true",
help="Keep extracted temporary files after completion",
default=False,
)
args = parser.parse_args()
DWARF2JSON = args.dwarfpath
@@ -132,4 +146,4 @@ if __name__ == '__main__':
urls += [[lines[2 * i].strip(), lines[(2 * i) + 1].strip()]]
d = Downloader(urls)
d.download_lists(keep = args.keep)
d.download_lists(keep=args.keep)
test/plugins/windows/test_scheduled_tasks.py
+2
View File
@@ -2,9 +2,11 @@ import sys
import struct
import traceback
import unittest
sys.path.insert(0, "../../volatility3")
from volatility3.plugins.windows import scheduled_tasks
class TestActionsDecoding(unittest.TestCase):
def test_decode_exe_action(self):
# fmt: off
volatility3/framework/layers/resources.py
+1 -1
View File
@@ -29,7 +29,7 @@ except ImportError:
try:
# Import so that the handler is found by the framework.class_subclasses callc
from smb import SMBHandler as SMBHandler # lgtm [py/unused-import]
from smb import SMBHandler as SMBHandler # lgtm [py/unused-import]
except ImportError:
# If we fail to import this, it means that SMB handling won't be available
pass