diff --git a/development/banner_server.py b/development/banner_server.py index 3aea41c82..b62477a26 100644 --- a/development/banner_server.py +++ b/development/banner_server.py @@ -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() diff --git a/development/compare-vol.py b/development/compare-vol.py index 1074c5d9c..a01d8e93c 100644 --- a/development/compare-vol.py +++ b/development/compare-vol.py @@ -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() diff --git a/development/mac-kdk/parse_pbzx2.py b/development/mac-kdk/parse_pbzx2.py index 1ca212211..b175539b3 100644 --- a/development/mac-kdk/parse_pbzx2.py +++ b/development/mac-kdk/parse_pbzx2.py @@ -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() diff --git a/development/pdbparse-to-json.py b/development/pdbparse-to-json.py index 6ffa4ea49..49b4da009 100644 --- a/development/pdbparse-to-json.py +++ b/development/pdbparse-to-json.py @@ -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}") diff --git a/development/schema_validate.py b/development/schema_validate.py index 0ea82d537..cf9565d68 100644 --- a/development/schema_validate.py +++ b/development/schema_validate.py @@ -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() diff --git a/development/stock-linux-json.py b/development/stock-linux-json.py index 877f78e1c..967cc7e18 100644 --- a/development/stock-linux-json.py +++ b/development/stock-linux-json.py @@ -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) diff --git a/test/plugins/windows/test_scheduled_tasks.py b/test/plugins/windows/test_scheduled_tasks.py index fdb19fbae..15d7f79a6 100644 --- a/test/plugins/windows/test_scheduled_tasks.py +++ b/test/plugins/windows/test_scheduled_tasks.py @@ -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 diff --git a/volatility3/framework/layers/resources.py b/volatility3/framework/layers/resources.py index 00215f624..236d256f1 100644 --- a/volatility3/framework/layers/resources.py +++ b/volatility3/framework/layers/resources.py @@ -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