Core: Apply black to the entire codebase and add workflow

This commit is contained in:
Mike Auty
2022-12-07 20:48:14 +00:00
parent 06ced0d59a
commit bd402772c7
189 changed files with 16494 additions and 8343 deletions
+63 -34
View File
@@ -26,18 +26,27 @@ class Bash(plugins.PluginInterface, timeliner.TimeLinerInterface):
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(name = 'kernel', description = 'Linux kernel',
architectures = ["Intel32", "Intel64"]),
requirements.PluginRequirement(name = 'pslist', plugin = pslist.PsList, version = (2, 0, 0)),
requirements.ListRequirement(name = 'pid',
element_type = int,
description = "Process IDs to include (all other processes are excluded)",
optional = True)
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(2, 0, 0)
),
requirements.ListRequirement(
name="pid",
element_type=int,
description="Process IDs to include (all other processes are excluded)",
optional=True,
),
]
def _generator(self, tasks):
vmlinux = self.context.modules[self.config["kernel"]]
is_32bit = not symbols.symbol_table_is_64bit(self.context, vmlinux.symbol_table_name)
is_32bit = not symbols.symbol_table_is_64bit(
self.context, vmlinux.symbol_table_name
)
if is_32bit:
pack_format = "I"
bash_json_file = "bash32"
@@ -45,10 +54,13 @@ 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)
@@ -64,44 +76,61 @@ 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 = []
if bang_addrs:
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)
for hist in sorted(history_entries, key = lambda x: x.get_time_as_integer()):
yield (0, (task.pid, task_name, hist.get_time_object(), hist.get_command()))
for hist in sorted(history_entries, key=lambda x: x.get_time_as_integer()):
yield (
0,
(task.pid, task_name, hist.get_time_object(), hist.get_command()),
)
def run(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
return renderers.TreeGrid([("PID", int), ("Process", str), ("CommandTime", datetime.datetime),
("Command", str)],
self._generator(
pslist.PsList.list_tasks(self.context,
self.config['kernel'],
filter_func = filter_func)))
return renderers.TreeGrid(
[
("PID", int),
("Process", str),
("CommandTime", datetime.datetime),
("Command", str),
],
self._generator(
pslist.PsList.list_tasks(
self.context, self.config["kernel"], filter_func=filter_func
)
),
)
def generate_timeline(self):
filter_func = pslist.PsList.create_pid_filter(self.config.get('pid', None))
filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None))
for row in self._generator(
pslist.PsList.list_tasks(self.context,
self.config['kernel'],
filter_func = filter_func)):
pslist.PsList.list_tasks(
self.context, self.config["kernel"], filter_func=filter_func
)
):
_depth, row_data = row
description = f"{row_data[0]} ({row_data[1]}): \"{row_data[3]}\""
description = f'{row_data[0]} ({row_data[1]}): "{row_data[3]}"'
yield (description, timeliner.TimeLinerType.CREATED, row_data[2])