Reformat all python files using yapf with custom volatility style.

This commit is contained in:
Mike Auty
2018-12-16 16:50:17 +00:00
parent 5bf2b78884
commit 19572b6e1e
99 changed files with 1672 additions and 2158 deletions
+24 -32
View File
@@ -21,11 +21,11 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def _generator(self, tasks):
is_32bit = not symbols.symbol_table_is_64bit(self.context, self.config["vmlinux"])
@@ -36,13 +36,10 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
pack_format = "Q"
bash_json_file = "bash64"
bash_table_name = BashIntermedSymbols.create(self.context,
self.config_path,
"linux",
bash_json_file)
bash_table_name = BashIntermedSymbols.create(self.context, self.config_path, "linux", bash_json_file)
ts_offset = self.context.symbol_space.get_type(
bash_table_name + constants.BANG + "hist_entry").relative_child_offset("timestamp")
ts_offset = self.context.symbol_space.get_type(bash_table_name + constants.BANG +
"hist_entry").relative_child_offset("timestamp")
for task in tasks:
task_name = utility.array_to_string(task.comm)
@@ -58,19 +55,22 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
bang_addrs = []
# find '#' values on the heap
for address in proc_layer.scan(self.context,
scanners.BytesScanner(b"#"),
sections = task.get_process_memory_sections(heap_only = True)):
for address in proc_layer.scan(
self.context,
scanners.BytesScanner(b"#"),
sections = task.get_process_memory_sections(heap_only = True)):
bang_addrs.append(struct.pack(pack_format, address))
history_entries = []
for address, _ in proc_layer.scan(self.context,
scanners.MultiStringScanner(bang_addrs),
sections = task.get_process_memory_sections(heap_only = True)):
hist = self.context.object(bash_table_name + constants.BANG + "hist_entry",
offset = address - ts_offset,
layer_name = proc_layer_name)
for address, _ in proc_layer.scan(
self.context,
scanners.MultiStringScanner(bang_addrs),
sections = task.get_process_memory_sections(heap_only = True)):
hist = self.context.object(
bash_table_name + constants.BANG + "hist_entry",
offset = address - ts_offset,
layer_name = proc_layer_name)
if hist.is_valid():
history_entries.append(hist)
@@ -84,24 +84,16 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
plugin = pslist.PsList.list_tasks
return renderers.TreeGrid(
[("PID", int),
("Process", str),
("CommandTime", datetime.datetime),
("Command", str)],
self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filt)))
[("PID", int), ("Process", str), ("CommandTime", datetime.datetime), ("Command", str)],
self._generator(plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filt)))
def generate_timeline(self):
filt = pslist.PsList.create_filter([self.config.get('pid', None)])
plugin = pslist.PsList.list_tasks
for row in self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filt)):
for row in self._generator(
plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filt)):
_depth, row_data = row
description = "{} ({}): \"{}\"".format(row_data[0], row_data[1], row_data[3])
yield (description, timeliner.TimeLinerType.CREATED, row_data[2])
@@ -19,11 +19,11 @@ class Check_afinfo(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
# returns whether the symbol is found within the kernel (system.map) or not
def _is_known_address(self, handler_addr):
@@ -62,9 +62,7 @@ class Check_afinfo(plugins.PluginInterface):
_, aslr_shift = linux.LinuxUtilities.find_aslr(self.context, self.config['vmlinux'], self.config['primary'])
vmlinux = self.context.module(self.config['vmlinux'], self.config['primary'], aslr_shift)
linux.LinuxUtilities.aslr_mask_symbol_table(self.context,
self.config['primary'],
self.config['vmlinux'],
linux.LinuxUtilities.aslr_mask_symbol_table(self.context, self.config['primary'], self.config['vmlinux'],
aslr_shift)
op_members = vmlinux.get_type('file_operations').members
@@ -89,8 +87,5 @@ class Check_afinfo(plugins.PluginInterface):
def run(self):
return renderers.TreeGrid(
[("Symbol Name", str),
("Member", str),
("Handler Address", format_hints.Hex)],
self._generator())
return renderers.TreeGrid([("Symbol Name", str), ("Member", str), ("Handler Address", format_hints.Hex)],
self._generator())
@@ -26,11 +26,11 @@ class Check_syscall(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def _get_table_size_next_symbol(self, table_addr, ptr_sz, vmlinux):
"""
@@ -94,7 +94,8 @@ class Check_syscall(plugins.PluginInterface):
md = capstone.Cs(capstone.CS_ARCH_X86, mode)
try:
func_addr = self.context.symbol_space.get_symbol(vmlinux.name + constants.BANG + syscall_entry_func).address
func_addr = self.context.symbol_space.get_symbol(vmlinux.name + constants.BANG +
syscall_entry_func).address
except exceptions.SymbolError as e:
# if we can't find the disassemble function then bail and rely on a different method
return 0
@@ -127,9 +128,7 @@ class Check_syscall(plugins.PluginInterface):
_, aslr_shift = linux.LinuxUtilities.find_aslr(self.context, self.config['vmlinux'], self.config['primary'])
vmlinux = self.context.module(self.config['vmlinux'], self.config['primary'], aslr_shift)
linux.LinuxUtilities.aslr_mask_symbol_table(self.context,
self.config['vmlinux'],
self.config['primary'],
linux.LinuxUtilities.aslr_mask_symbol_table(self.context, self.config['vmlinux'], self.config['primary'],
aslr_shift)
ptr_sz = vmlinux.get_type("pointer").size
@@ -159,8 +158,8 @@ class Check_syscall(plugins.PluginInterface):
tables.append(("32bit", ia32_info))
for (table_name, (tableaddr, tblsz)) in tables:
table = vmlinux.object(type_name = "array", subtype = vmlinux.get_type("pointer"),
offset = tableaddr, count = tblsz)
table = vmlinux.object(
type_name = "array", subtype = vmlinux.get_type("pointer"), offset = tableaddr, count = tblsz)
for (i, call_addr) in enumerate(table):
if not call_addr:
@@ -178,10 +177,5 @@ class Check_syscall(plugins.PluginInterface):
def run(self):
return renderers.TreeGrid(
[("Table Address", format_hints.Hex),
("Table Name", str),
("Index", int),
("Handler Address", format_hints.Hex),
("Handler Symbol", str)],
self._generator())
return renderers.TreeGrid([("Table Address", format_hints.Hex), ("Table Name", str), ("Index", int),
("Handler Address", format_hints.Hex), ("Handler Symbol", str)], self._generator())
+8 -21
View File
@@ -17,11 +17,11 @@ class Elfs(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def _generator(self, tasks):
for task in tasks:
@@ -40,14 +40,7 @@ class Elfs(plugins.PluginInterface):
path = vma.get_name(task)
yield (
0,
(task.pid,
name,
format_hints.Hex(vma.vm_start),
format_hints.Hex(vma.vm_end),
path
))
yield (0, (task.pid, name, format_hints.Hex(vma.vm_start), format_hints.Hex(vma.vm_end), path))
def run(self):
filt = pslist.PsList.create_filter([self.config.get('pid', None)])
@@ -55,12 +48,6 @@ class Elfs(plugins.PluginInterface):
plugin = pslist.PsList.list_tasks
return renderers.TreeGrid(
[("PID", int),
("Process", str),
("Start", format_hints.Hex),
("End", format_hints.Hex),
[("PID", int), ("Process", str), ("Start", format_hints.Hex), ("End", format_hints.Hex),
("File Path", str)],
self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filt)))
self._generator(plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filt)))
+8 -17
View File
@@ -17,17 +17,14 @@ class Lsmod(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
@classmethod
def list_modules(cls,
context: interfaces.context.ContextInterface,
layer_name: str,
vmlinux_symbols: str):
def list_modules(cls, context: interfaces.context.ContextInterface, layer_name: str, vmlinux_symbols: str):
"""Lists all the modules in the primary layer"""
_, aslr_shift = linux.LinuxUtilities.find_aslr(context, vmlinux_symbols, layer_name)
@@ -43,9 +40,7 @@ class Lsmod(plugins.PluginInterface):
yield module
def _generator(self):
for module in self.list_modules(self.context,
self.config['primary'],
self.config['vmlinux']):
for module in self.list_modules(self.context, self.config['primary'], self.config['vmlinux']):
mod_size = module.get_init_size() + module.get_core_size()
@@ -54,8 +49,4 @@ class Lsmod(plugins.PluginInterface):
yield 0, (format_hints.Hex(module.vol.offset), mod_name, mod_size)
def run(self):
return renderers.TreeGrid(
[("Offset", format_hints.Hex),
("Name", str),
("Size", int)],
self._generator())
return renderers.TreeGrid([("Offset", format_hints.Hex), ("Name", str), ("Size", int)], self._generator())
+9 -15
View File
@@ -19,19 +19,19 @@ class Lsof(plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def _generator(self, tasks):
for task in tasks:
name = utility.array_to_string(task.comm)
pid = int(task.pid)
for fd_num, _, full_path in linux.LinuxUtilities.files_descriptors_for_process(self.config, self.context,
task):
for fd_num, _, full_path in linux.LinuxUtilities.files_descriptors_for_process(
self.config, self.context, task):
yield (0, (pid, name, fd_num, full_path))
def run(self):
@@ -42,11 +42,5 @@ class Lsof(plugins.PluginInterface):
plugin = pslist.PsList.list_tasks
return renderers.TreeGrid(
[("PID", int),
("Process", str),
("FD", int),
("Path", str)],
self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filter)))
[("PID", int), ("Process", str), ("FD", int), ("Path", str)],
self._generator(plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filter)))
+11 -23
View File
@@ -15,11 +15,11 @@ class Malfind(interfaces_plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def list_injections(self, task):
"""Generate memory regions for a process that may contain
@@ -55,27 +55,15 @@ class Malfind(interfaces_plugins.PluginInterface):
disasm = interfaces_renderers.Disassembly(data, vma.vm_start, architecture)
yield (0, (task.pid,
process_name,
format_hints.Hex(vma.vm_start),
format_hints.Hex(vma.vm_end),
vma.get_protection(),
format_hints.HexBytes(data),
disasm))
yield (0, (task.pid, process_name, format_hints.Hex(vma.vm_start), format_hints.Hex(vma.vm_end),
vma.get_protection(), format_hints.HexBytes(data), disasm))
def run(self):
filt = pslist.PsList.create_filter([self.config.get('pid', None)])
plugin = pslist.PsList.list_tasks
return renderers.TreeGrid([("PID", int),
("Process", str),
("Start", format_hints.Hex),
("End", format_hints.Hex),
("Protection", str),
("Hexdump", format_hints.HexBytes),
("Disasm", interfaces_renderers.Disassembly)],
self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filt)))
return renderers.TreeGrid(
[("PID", int), ("Process", str), ("Start", format_hints.Hex), ("End", format_hints.Hex),
("Protection", str), ("Hexdump", format_hints.HexBytes), ("Disasm", interfaces_renderers.Disassembly)],
self._generator(plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filt)))
+10 -32
View File
@@ -16,11 +16,11 @@ class Maps(plugins.PluginInterface):
@classmethod
def get_requirements(cls):
# Since we're calling the plugin, make sure we have the plugin's requirements
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
def _generator(self, tasks):
for task in tasks:
@@ -46,19 +46,8 @@ class Maps(plugins.PluginInterface):
path = vma.get_name(task)
yield (
0,
(task.pid,
name,
format_hints.Hex(vma.vm_start),
format_hints.Hex(vma.vm_end),
flags,
format_hints.Hex(page_offset),
major,
minor,
inode,
path
))
yield (0, (task.pid, name, format_hints.Hex(vma.vm_start), format_hints.Hex(vma.vm_end), flags,
format_hints.Hex(page_offset), major, minor, inode, path))
def run(self):
filter = pslist.PsList.create_filter([self.config.get('pid', None)])
@@ -66,17 +55,6 @@ class Maps(plugins.PluginInterface):
plugin = pslist.PsList.list_tasks
return renderers.TreeGrid(
[("PID", int),
("Process", str),
("Start", format_hints.Hex),
("End", format_hints.Hex),
("Flags", str),
("PgOff", format_hints.Hex),
("Major", int),
("Minor", int),
("Inode", int),
("File Path", str)],
self._generator(plugin(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = filter)))
[("PID", int), ("Process", str), ("Start", format_hints.Hex), ("End", format_hints.Hex), ("Flags", str),
("PgOff", format_hints.Hex), ("Major", int), ("Minor", int), ("Inode", int), ("File Path", str)],
self._generator(plugin(self.context, self.config['primary'], self.config['vmlinux'], filter = filter)))
+12 -14
View File
@@ -12,11 +12,11 @@ class PsList(interfaces_plugins.PluginInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [requirements.TranslationLayerRequirement(name = 'primary',
description = 'Kernel Address Space',
architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux",
description = "Linux Kernel")]
return [
requirements.TranslationLayerRequirement(
name = 'primary', description = 'Kernel Address Space', architectures = ["Intel32", "Intel64"]),
requirements.SymbolRequirement(name = "vmlinux", description = "Linux Kernel")
]
@classmethod
def create_filter(cls, pid_list: List[int] = None) -> Callable[[int], bool]:
@@ -24,6 +24,7 @@ class PsList(interfaces_plugins.PluginInterface):
pid_list = pid_list or []
filter_list = [x for x in pid_list if x is not None]
if filter_list:
def filter_func(x):
return x not in filter_list
@@ -32,10 +33,11 @@ class PsList(interfaces_plugins.PluginInterface):
return lambda _: False
def _generator(self):
for task in self.list_tasks(self.context,
self.config['primary'],
self.config['vmlinux'],
filter = self.create_filter([self.config.get('pid', None)])):
for task in self.list_tasks(
self.context,
self.config['primary'],
self.config['vmlinux'],
filter = self.create_filter([self.config.get('pid', None)])):
pid = task.pid
ppid = 0
if task.parent:
@@ -49,7 +51,6 @@ class PsList(interfaces_plugins.PluginInterface):
layer_name: str,
vmlinux_symbols: str,
filter: Callable[[int], bool] = lambda _: False) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the tasks in the primary layer"""
_, aslr_shift = linux.LinuxUtilities.find_aslr(context, vmlinux_symbols, layer_name)
@@ -61,7 +62,4 @@ class PsList(interfaces_plugins.PluginInterface):
yield task
def run(self):
return renderers.TreeGrid([("PID", int),
("PPID", int),
("COMM", str)],
self._generator())
return renderers.TreeGrid([("PID", int), ("PPID", int), ("COMM", str)], self._generator())
+1 -3
View File
@@ -38,9 +38,7 @@ class PsTree(pslist.PsList):
def yield_processes(pid):
proc = self._processes[pid]
row = (proc.pid,
proc.parent.pid,
utility.array_to_string(proc.comm))
row = (proc.pid, proc.parent.pid, utility.array_to_string(proc.comm))
yield (self._levels[pid] - 1, row)
for child_pid in self._children.get(pid, []):