Apply 'Black' suggestions

This commit is contained in:
Gustavo Moreira
2023-05-08 11:54:48 +02:00
parent a09f77897b
commit ced6ff346c
4 changed files with 37 additions and 19 deletions
@@ -141,16 +141,18 @@ class MountInfo(plugins.PluginInterface):
)
def _get_tasks_mountpoints(
self, tasks: Iterable[interfaces.objects.ObjectInterface], filtered_by_pids: bool
self,
tasks: Iterable[interfaces.objects.ObjectInterface],
filtered_by_pids: bool,
):
seen_mountpoints = set()
for task in tasks:
if not (
task and
task.fs and
task.fs.root and
task.nsproxy and
task.nsproxy.mnt_ns
task
and task.fs
and task.fs.root
and task.nsproxy
and task.nsproxy.mnt_ns
):
# This task doesn't have all the information required.
# It should be a kernel < 2.6.30
@@ -183,19 +185,23 @@ class MountInfo(plugins.PluginInterface):
filtered_by_pids: bool,
) -> Iterable[Tuple[int, Tuple]]:
warning_shown = False
for task, mnt, mnt_ns_id in self._get_tasks_mountpoints(tasks, filtered_by_pids):
for task, mnt, mnt_ns_id in self._get_tasks_mountpoints(
tasks, filtered_by_pids
):
if (
not warning_shown and
mnt_ns_ids and
isinstance(mnt_ns_id, renderers.NotAvailableValue)
not warning_shown
and mnt_ns_ids
and isinstance(mnt_ns_id, renderers.NotAvailableValue)
):
vollog.warning("Cannot filter by namespace id, it is not available in this kernel.")
vollog.warning(
"Cannot filter by namespace id, it is not available in this kernel."
)
warning_shown = True
if (
not isinstance(mnt_ns_id, renderers.NotAvailableValue) and
mnt_ns_ids and
mnt_ns_id not in mnt_ns_ids
not isinstance(mnt_ns_id, renderers.NotAvailableValue)
and mnt_ns_ids
and mnt_ns_id not in mnt_ns_ids
):
continue
@@ -284,5 +290,6 @@ class MountInfo(plugins.PluginInterface):
columns.extend(extra_columns)
return renderers.TreeGrid(
columns, self._generator(tasks, mount_ns_ids, mount_format, filtered_by_pids)
columns,
self._generator(tasks, mount_ns_ids, mount_format, filtered_by_pids),
)
@@ -65,7 +65,10 @@ class SockHandlers(interfaces.configuration.VersionableInterface):
self._vmlinux.symbol_table_name + constants.BANG + "net_device"
)
for net_dev in net.dev_base_head.to_list(net_device_symname, "dev_list"):
if isinstance(netns_id, NotAvailableValue) or net.get_inode() != netns_id:
if (
isinstance(netns_id, NotAvailableValue)
or net.get_inode() != netns_id
):
continue
dev_name = utility.array_to_string(net_dev.name)
netdevices_map[net_dev.ifindex] = dev_name
@@ -392,7 +392,9 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
symbol_table_arr = volobj.vol.type_name.split("!", 1)
symbol_table = symbol_table_arr[0] if len(symbol_table_arr) == 2 else None
module_names = volobj._context.modules.get_modules_by_symbol_tables(symbol_table)
module_names = volobj._context.modules.get_modules_by_symbol_tables(
symbol_table
)
module_names = list(module_names)
if not module_names:
@@ -878,7 +878,9 @@ class vfsmount(objects.StructType):
if type(vfsmount_ptr) == objects.Pointer:
return self.vol.offset == vfsmount_ptr
else:
raise exceptions.VolatilityException("Unexpected argument type. It has to be a 'vfsmount *'")
raise exceptions.VolatilityException(
"Unexpected argument type. It has to be a 'vfsmount *'"
)
def _get_real_mnt(self):
"""Gets the struct 'mount' containing this 'vfsmount'.
@@ -889,7 +891,9 @@ class vfsmount(objects.StructType):
mount: the struct 'mount' containing this 'vfsmount'.
"""
vmlinux = linux.LinuxUtilities.get_vmlinux_from_volobj(self)
return linux.LinuxUtilities.container_of(self.vol.offset, "mount", "mnt", vmlinux)
return linux.LinuxUtilities.container_of(
self.vol.offset, "mount", "mnt", vmlinux
)
def get_vfsmnt_current(self):
"""Returns the current fs where we are mounted on
@@ -994,6 +998,7 @@ class vfsmount(objects.StructType):
def get_devname(self) -> str:
return utility.pointer_to_string(self.mnt_devname, count=255)
class kobject(objects.StructType):
def reference_count(self):
refcnt = self.kref.refcount
@@ -1262,6 +1267,7 @@ class netlink_sock(objects.StructType):
else:
raise AttributeError("Unable to find a destination port id")
class vsock_sock(objects.StructType):
def get_protocol(self):
# The protocol should always be 0 for vsocks