Update to using str dictionary key. Add requirements for each gatherer in modxview. Validate names are unique while sanity checking.

This commit is contained in:
Andrew Case
2025-03-15 17:22:31 +00:00
parent 53e32e36f6
commit 674cc045d2
2 changed files with 40 additions and 14 deletions
@@ -38,7 +38,17 @@ spot modules presence and taints."""
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGatherers,
component=linux_utilities_modules.ModuleGathererLsmod,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGathererSysFs,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGathererScanner,
version=(1, 0, 0),
),
requirements.VersionRequirement(
@@ -113,14 +123,14 @@ spot modules presence and taints."""
# We want to be explicit on the plugins results we are interested in
for gatherer in wanted_gatherers:
# Iterate over each recovered module
for mod_info in run_results[gatherer]:
for mod_info in run_results[gatherer.name]:
# Use offsets as unique keys, whether a module
# appears in many plugin runs or not
if aggregated_modules.get(mod_info.offset, None) is not None:
# Append the plugin to the list of originating plugins
aggregated_modules[mod_info.offset].append(gatherer)
aggregated_modules[mod_info.offset].append(gatherer.name)
else:
aggregated_modules[mod_info.offset] = [gatherer]
aggregated_modules[mod_info.offset] = [gatherer.name]
for module_offset, gatherers in aggregated_modules.items():
module = kernel.object("module", offset=module_offset, absolute=True)
@@ -148,9 +158,9 @@ spot modules presence and taints."""
(
module.get_name() or NotAvailableValue(),
format_hints.Hex(module_offset),
linux_utilities_modules.ModuleGathererLsmod in gatherers,
linux_utilities_modules.ModuleGathererSysFs in gatherers,
linux_utilities_modules.ModuleGathererScanner in gatherers,
linux_utilities_modules.ModuleGathererLsmod.name in gatherers,
linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
linux_utilities_modules.ModuleGathererScanner.name in gatherers,
taints or NotAvailableValue(),
),
)
@@ -233,19 +233,21 @@ class Modules(interfaces.configuration.VersionableInterface):
kernel_module_name: str,
caller_wanted_gatherers: List[ModuleGathererInterface],
flatten: bool = True,
) -> Dict[ModuleGathererInterface, List[ModuleInfo]]:
) -> Dict[str, List[ModuleInfo]]:
"""Run module scanning plugins and aggregate the results. It is designed
to not operate any inter-plugin results triage.
Rules for `caller_wanted_sources`:
Rules for `caller_wanted_gatherers`:
If `ModuleGatherers.all_gathers_identifier` is specified then every source will be populated
If empty or an invalid source is specified then a ValueError is thrown
If empty or an invalid gatherer is specified then a ValueError is thrown
All gatherer names must be unique
Args:
called_wanted_sources: The list of sources to gather modules.
flatten: Whether to de-duplicate modules across sources
flatten: Whether to de-duplicate modules across gatherers
Returns:
Dictionary mapping each plugin to its corresponding result
Dictionary mapping each gatherer to its corresponding result
"""
if not caller_wanted_gatherers:
raise ValueError(
@@ -255,12 +257,26 @@ class Modules(interfaces.configuration.VersionableInterface):
if not isinstance(caller_wanted_gatherers, Iterable):
raise ValueError("`caller_wanted_gatherers` must be iterable")
seen_names = set()
for gatherer in caller_wanted_gatherers:
if not issubclass(gatherer, ModuleGathererInterface):
raise ValueError(
f"Invalid gatherer sent through `caller_wanted_gatherers`: {gatherer}"
)
if not hasattr(gatherer, "name"):
raise ValueError(
f"{gatherer} does not have a name attribute, which is required."
)
if gatherer.name in seen_names:
raise ValueError(
f"{gatherer} has a name {gatherer.name} which has already been processed. Names must be unique."
)
seen_names.add(gatherer.name)
kernel = context.modules[kernel_module_name]
address_mask = context.layers[kernel.layer_name].address_mask
@@ -269,7 +285,7 @@ class Modules(interfaces.configuration.VersionableInterface):
# Walk each source gathering modules
for gatherer in caller_wanted_gatherers:
run_results[gatherer] = []
run_results[gatherer.name] = []
# process each module coming from back the current source
for module in gatherer.gather_modules(context, kernel_module_name):
@@ -281,7 +297,7 @@ class Modules(interfaces.configuration.VersionableInterface):
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
run_results[gatherer].append(modinfo)
run_results[gatherer.name].append(modinfo)
if flatten:
return cls.flatten_run_modules_results(run_results)