Add output_def to PluginFunction

dissect/target/plugin.py

@@ -1077,6 +1077,7 @@ class PluginFunction:
  name: str
  path: str
  output_type: str
+ output_def: Optional[RecordDescriptor]
  class_object: type[Plugin]
  method_name: str
  plugin_desc: PluginDescriptor = field(hash=False)
@@ -1236,6 +1237,7 @@ def find_plugin_functions(
  class_object=loaded_plugin_object,
  method_name=method_name,
  output_type=getattr(fobject, "__output__", "text"),
+ output_def=getattr(fobject, "__record__", None),
  plugin_desc=func,
  )
  )
@@ -1281,6 +1283,7 @@ def find_plugin_functions(
  class_object=loaded_plugin_object,
  method_name=funcname,
  output_type=getattr(fobject, "__output__", "text"),
+ output_def=getattr(fobject, "__record__", None),
  plugin_desc=description,
  )
  )

tests/test_plugin.py

@@ -5,7 +5,7 @@
 from unittest.mock import Mock, patch
 
 import pytest
-from flow.record import Record
+from flow.record import Record, RecordDescriptor, RecordField
 
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.descriptor_extensions import UserRecordDescriptorExtension
@@ -15,6 +15,7 @@
  NamespacePlugin,
  OSPlugin,
  Plugin,
+ PluginFunction,
  environment_variable_paths,
  export,
  find_plugin_functions,
@@ -139,6 +140,43 @@ def test_find_plugin_function_linux(target_linux: Target) -> None:
  assert found[0].path == "os.unix.linux.services.services"
 
 
+def test_find_plugin_function_record_def(target_linux: Target) -> None:
+ """test if found plugins have an output_def attribute with a RecordDescriptor inside"""
+
+ # Plugins yielding records should have a RecordDescriptor in the output_def attribute.
+ found, _ = find_plugin_functions(target_linux, "users")
+ assert len(found) == 1
+ users_plugin = found[0]
+ assert users_plugin.output_type == "record"
+ assert hasattr(users_plugin, "output_def")
+ assert isinstance(users_plugin.output_def, RecordDescriptor)
+ assert hasattr(users_plugin.output_def, "fields")
+ assert isinstance(users_plugin.output_def.fields, dict)
+ assert isinstance(users_plugin.output_def.fields.get("hostname"), RecordField)
+
+ # Plugins yielding non-records should have a falsy output_def attribute.
+ found, _ = find_plugin_functions(target_linux, "os")
+ assert len(found) == 1
+ os_plugin = found[0]
+ assert os_plugin.output_type == "text"
+ assert os_plugin.output_def is None
+
+
+@pytest.mark.parametrize(
+ "func_path, func",
+ [
+ (func.path, func)
+ for func in find_plugin_functions(Target(), "*", compatibility=False, show_hidden=True)[0]
+ if func.output_type == "record"
+ ],
+)
+def test_exported_plugin_record_def(func_path: str, func: PluginFunction) -> None:
+ """test if every exported plugin yielding records has defined a RecordDescriptor"""
+ assert isinstance(func.output_def, RecordDescriptor) or (
+ isinstance(func.output_def, list) and all([isinstance(f, RecordDescriptor) for f in func.output_def])
+ )
+
+
 TestRecord = create_extended_descriptor([UserRecordDescriptorExtension])(
  "application/test",
  [

tests/tools/test_query.py

@@ -160,6 +160,7 @@ def mock_find_plugin_function(
  PluginFunction(
  name=pattern,
  output_type="record",
+ output_def=None,
  path=pattern,
  class_object=MagicMock(),
  method_name=pattern,