Add NamespacePlugin (DIS-1681)

dissect/target/plugin.py

@@ -793,6 +793,48 @@ def _modulepath(cls) -> str:
     return cls.__module__.replace(MODULE_PATH, "").lstrip(".")
 
 
+class NamespacePlugin(Plugin):
+    __findable__ = False
+
+    def __init__(self, target: Target):
+        """A Namespace plugin provides services to access functionality
+        from a group of sub plugins through an internal function _func.
+        Upon initialisation, sub plugins are collected.
+        """
+        super().__init__(target)
+        self._plugins = []
+
+        for attr in ["__namespace__", "SUBPLUGINS"]:
+            if not hasattr(self, attr):
+                raise PluginError(f"Namespace plugin lacks {attr} attribute")
+
+        for entry in self.SUBPLUGINS:
+            try:
+                self._plugins.append(getattr(self.target, entry))
+            except Exception:  # noqa
+                target.log.exception("Failed to load sub plugin: %s", entry)
+
+    def _func(self, func_name: str) -> None:
+        """Return the supported sub plugin records.
+
+        Args:
+            func_name: Exported function of the sub plugin to find.
+
+        Yields:
+            Record from the sub function.
+        """
+        for p in self._plugins:
+            try:
+                for entry in getattr(p, func_name)():
+                    yield entry
+            except Exception:
+                self.target.log.exception("Failed to execute sub plugin: {}.{}", p._name, func_name)
+
+    def check_compatible(self):
+        if not len(self._plugins):
+            raise UnsupportedPluginError("No compatible sub plugins found")
+
+
 # Needs to be at the bottom of the module because __init_subclass__ requires everything
 # in the parent class Plugin to be defined and resolved.
 class InternalPlugin(Plugin):

dissect/target/plugins/apps/remoteaccess/remoteaccess.py

@@ -1,7 +1,7 @@
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.descriptor_extensions import UserRecordDescriptorExtension
 from dissect.target.helpers.record import create_extended_descriptor
-from dissect.target.plugin import Plugin, export
+from dissect.target.plugin import NamespacePlugin, export
 
 RemoteAccessRecord = create_extended_descriptor([UserRecordDescriptorExtension])(
     "application/log/remoteaccess",
@@ -14,7 +14,7 @@
 )
 
 
-class RemoteAccessPlugin(Plugin):
+class RemoteAccessPlugin(NamespacePlugin):
     """General Remote Access plugin.
 
     This plugin groups the functions of all remote access plugins. For example,
@@ -23,34 +23,12 @@ class RemoteAccessPlugin(Plugin):
     """
 
     __namespace__ = "remoteaccess"
-    __findable__ = False
 
-    TOOLS = [
+    SUBPLUGINS = [
         "teamviewer",
         "anydesk",
     ]
 
-    def __init__(self, target):
-        super().__init__(target)
-        self._plugins = []
-        for entry in self.TOOLS:
-            try:
-                self._plugins.append(getattr(self.target, entry))
-            except Exception:  # noqa
-                target.log.exception("Failed to load tool plugin: %s", entry)
-
-    def check_compatible(self):
-        if not len(self._plugins):
-            raise UnsupportedPluginError("No compatible tool plugins found")
-
-    def _func(self, f):
-        for p in self._plugins:
-            try:
-                for entry in getattr(p, f)():
-                    yield entry
-            except Exception:
-                self.target.log.exception("Failed to execute tool plugin: {}.{}", p._name, f)
-
     @export(record=RemoteAccessRecord)
     def remoteaccess(self):
         """Return Remote Access records from all Remote Access Tools.

dissect/target/plugins/browsers/browser.py

@@ -1,7 +1,7 @@
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.descriptor_extensions import UserRecordDescriptorExtension
 from dissect.target.helpers.record import create_extended_descriptor
-from dissect.target.plugin import Plugin, export
+from dissect.target.plugin import NamespacePlugin, export
 
 GENERIC_DOWNLOAD_RECORD_FIELDS = [
     ("datetime", "ts_start"),
@@ -60,7 +60,7 @@
 )
 
 
-class BrowserPlugin(Plugin):
+class BrowserPlugin(NamespacePlugin):
     """General browser plugin.
 
     This plugin groups the functions of all browser plugins. For example,
@@ -69,45 +69,15 @@ class BrowserPlugin(Plugin):
     """
 
     __namespace__ = "browser"
-    __findable__ = False
 
-    BROWSERS = [
+    SUBPLUGINS = [
         "chrome",
         "chromium",
         "edge",
         "firefox",
         "iexplore",
     ]
 
-    def __init__(self, target):
-        super().__init__(target)
-        self._plugins = []
-        for entry in self.BROWSERS:
-            try:
-                self._plugins.append(getattr(self.target, entry))
-            except Exception:
-                target.log.exception("Failed to load browser plugin: %s", entry)
-
-    def _func(self, func_name: str):
-        """Return the supported browser plugin records.
-
-        Args:
-            func_name: Exported function of the browser plugin to find.
-
-        Yields:
-            Record from the browser function.
-        """
-        for plugin_name in self._plugins:
-            try:
-                for entry in getattr(plugin_name, func_name)():
-                    yield entry
-            except Exception:
-                self.target.log.exception("Failed to execute browser plugin: %s.%s", plugin_name._name, func_name)
-
-    def check_compatible(self) -> bool:
-        if not len(self._plugins):
-            raise UnsupportedPluginError("No compatible browser plugins found")
-
     @export(record=BrowserDownloadRecord)
     def downloads(self):
         """Return browser download records from all browsers installed and supported.

dissect/target/tools/query.py

@@ -148,7 +148,6 @@ def main():
     # search pattern, only display plugins that can be applied to ANY targets
     if args.list:
         collected_plugins = {}
-
         if args.targets:
             for target in args.targets:
                 plugin_target = Target.open(target)

tests/test_plugin.py

@@ -5,11 +5,17 @@
 import pytest
 
 from dissect.target.plugin import (
+    PLUGINS,
+    NamespacePlugin,
+    Plugin,
+    PluginError,
     environment_variable_paths,
+    export,
     find_plugin_functions,
     get_external_module_paths,
     save_plugin_import_failure,
 )
+from dissect.target.target import Target
 
 
 def test_save_plugin_import_failure():
@@ -118,3 +124,54 @@ def test_find_plugin_function_unix(target_unix):
 
     assert len(found) == 1
     assert found[0].name == "os.unix.services.services"
+
+
+def test_namespace_plugin(target_win:Target) -> None:
+    class TestPlugin(NamespacePlugin):
+        pass
+
+    class TestPlugin1(NamespacePlugin):
+        __namespace__ = "NS"
+
+    class TestPlugin2(NamespacePlugin):
+        __namespace__ = "NS"
+        __findable__ = False
+        SUBPLUGINS = ["subtest"]
+
+        @export(output="yield")
+        def test(self):
+            yield from self._func("test")
+
+    class Subtest(Plugin):
+        __namespace__ = "subtest"
+        __findable__ = False
+
+        @export(output="yield")
+        def test(self):
+            yield "test"
+
+    expected_error = None
+    try:
+        TestPlugin(target_win)
+    except PluginError as err:
+        expected_error = err
+    assert isinstance(expected_error, PluginError)
+
+    expected_error = None
+    try:
+        TestPlugin1(target_win)
+    except PluginError as err:
+        expected_error = err
+    assert isinstance(expected_error, PluginError)
+
+    expected_error = None
+    test_plugin = TestPlugin2(target_win)
+    assert isinstance(test_plugin, TestPlugin2)
+
+    target_win._register_plugin_functions(Subtest(target_win))
+    target_win._register_plugin_functions(TestPlugin2(target_win))
+
+    assert list(target_win.NS.test())[0] == "test"
+
+    # Remove test plugin from list afterwards to avoid order effects
+    del PLUGINS["tests"]