mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-09-28 12:34:53 +02:00
Development: Update development scripts with f-strings
This commit is contained in:
@@ -46,7 +46,7 @@ class VolatilityTest:
|
||||
self.create_prerequisites(plugin, image, image_hash)
|
||||
|
||||
# Volatility 2 Test
|
||||
print("[*] Testing {} {} with image {}".format(self.short_name, plugin.name, 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()
|
||||
@@ -56,9 +56,9 @@ class VolatilityTest:
|
||||
completed = excp
|
||||
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))
|
||||
print(f" Tested {self.short_name} {plugin.name} with image {image.filepath}: {total_time}")
|
||||
with open(
|
||||
os.path.join(self.output_directory, '{}_{}_{}_stdout'.format(self.short_name, plugin.name, image_hash)),
|
||||
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:
|
||||
@@ -91,15 +91,15 @@ class Volatility2Test(VolatilityTest):
|
||||
def create_prerequisites(self, plugin: VolatilityPlugin, image: VolatilityImage, image_hash):
|
||||
# Volatility 2 image info
|
||||
if not image.vol2_profile:
|
||||
print("[*] Testing {} imageinfo with image {}".format(self.short_name, 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)
|
||||
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))
|
||||
with open(os.path.join(self.output_directory, 'vol2_imageinfo_{}_stdout'.format(image_hash)), "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(b"Suggested Profile\(s\) : ([^,]+)", vol2_completed.stdout)[1]
|
||||
|
||||
|
||||
@@ -27,17 +27,17 @@ class PDBRetreiver:
|
||||
logger.info("Download PDB file...")
|
||||
file_name = ".".join(file_name.split(".")[:-1] + ['pdb'])
|
||||
for sym_url in ['http://msdl.microsoft.com/download/symbols']:
|
||||
url = sym_url + "/{}/{}/".format(file_name, guid)
|
||||
url = sym_url + f"/{file_name}/{guid}/"
|
||||
|
||||
result = None
|
||||
for suffix in [file_name[:-1] + '_', file_name]:
|
||||
try:
|
||||
logger.debug("Attempting to retrieve {}".format(url + suffix))
|
||||
logger.debug(f"Attempting to retrieve {url + suffix}")
|
||||
result, _ = request.urlretrieve(url + suffix)
|
||||
except request.HTTPError as excp:
|
||||
logger.debug("Failed with {}".format(excp))
|
||||
logger.debug(f"Failed with {excp}")
|
||||
if result:
|
||||
logger.debug("Successfully written to {}".format(result))
|
||||
logger.debug(f"Successfully written to {result}")
|
||||
break
|
||||
return result
|
||||
|
||||
@@ -257,7 +257,7 @@ class PDBConvertor:
|
||||
if output is None:
|
||||
import pdb
|
||||
pdb.set_trace()
|
||||
raise ValueError("Unknown size for field: {}".format(field.name))
|
||||
raise ValueError(f"Unknown size for field: {field.name}")
|
||||
return output
|
||||
|
||||
def _format_kind(self, kind):
|
||||
@@ -355,6 +355,6 @@ if __name__ == '__main__':
|
||||
json.dump(convertor.read_pdb(), f, indent = 2, sort_keys = True)
|
||||
|
||||
if args.keep:
|
||||
print("Temporary PDB file: {}".format(filename))
|
||||
print(f"Temporary PDB file: {filename}")
|
||||
elif delfile:
|
||||
os.remove(filename)
|
||||
|
||||
@@ -35,7 +35,7 @@ if __name__ == '__main__':
|
||||
for filename in args.filenames:
|
||||
try:
|
||||
if os.path.exists(filename):
|
||||
print("[?] Validating file: {}".format(filename))
|
||||
print(f"[?] Validating file: {filename}")
|
||||
with open(filename, 'r') as t:
|
||||
test = json.load(t)
|
||||
|
||||
@@ -45,14 +45,14 @@ if __name__ == '__main__':
|
||||
result = schemas.validate(test, False)
|
||||
|
||||
if result:
|
||||
print("[+] Validation successful: {}".format(filename))
|
||||
print(f"[+] Validation successful: {filename}")
|
||||
else:
|
||||
print("[-] Validation failed: {}".format(filename))
|
||||
print(f"[-] Validation failed: {filename}")
|
||||
failures.append(filename)
|
||||
else:
|
||||
print("[x] File not found: {}".format(filename))
|
||||
print(f"[x] File not found: {filename}")
|
||||
except Exception as e:
|
||||
failures.append(filename)
|
||||
print("[x] Exception occurred: {} ({})".format(filename, repr(e)))
|
||||
print(f"[x] Exception occurred: {filename} ({repr(e)})")
|
||||
|
||||
print("Failures", failures)
|
||||
|
||||
@@ -30,7 +30,7 @@ class Downloader:
|
||||
def download_list(self, urls: List[str]) -> Dict[str, str]:
|
||||
processed_files = {}
|
||||
for url in urls:
|
||||
print(" - Downloading {}".format(url))
|
||||
print(f" - Downloading {url}")
|
||||
data = requests.get(url)
|
||||
with tempfile.NamedTemporaryFile() as archivedata:
|
||||
archivedata.write(data.content)
|
||||
@@ -48,14 +48,14 @@ class Downloader:
|
||||
extracted = None
|
||||
for member in rpm.getmembers():
|
||||
if 'vmlinux' in member.name or 'System.map' in member.name:
|
||||
print(" - Extracting {}".format(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:
|
||||
print(" - Writing to {}".format(output.name))
|
||||
print(f" - Writing to {output.name}")
|
||||
output.write(extracted.read())
|
||||
return output.name
|
||||
|
||||
@@ -65,14 +65,14 @@ class Downloader:
|
||||
extracted = None
|
||||
for member in deb.data.tgz().getmembers():
|
||||
if member.name.endswith('vmlinux') or 'System.map' in member.name:
|
||||
print(" - Extracting {}".format(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:
|
||||
print(" - Writing to {}".format(output.name))
|
||||
print(f" - Writing to {output.name}")
|
||||
output.write(extracted.read())
|
||||
return output.name
|
||||
|
||||
@@ -81,7 +81,7 @@ class Downloader:
|
||||
print("Processing Files...")
|
||||
for i in named_files:
|
||||
if named_files[i] is None:
|
||||
print("FAILURE: None encountered for {}".format(i))
|
||||
print(f"FAILURE: None encountered for {i}")
|
||||
return
|
||||
args = [DWARF2JSON, 'linux']
|
||||
output_filename = 'unknown-kernel.json'
|
||||
@@ -91,10 +91,10 @@ class Downloader:
|
||||
prefix = '--elf'
|
||||
output_filename = './' + '-'.join((named_file.split('/')[-1]).split('-')[2:])[:-4] + '.json.xz'
|
||||
args += [prefix, named_files[named_file]]
|
||||
print(" - Running {}".format(args))
|
||||
print(f" - Running {args}")
|
||||
proc = subprocess.run(args, capture_output = True)
|
||||
|
||||
print(" - Writing to {}".format(output_filename))
|
||||
print(f" - Writing to {output_filename}")
|
||||
with lzma.open(output_filename, 'w') as f:
|
||||
f.write(proc.stdout)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user