Merge remote-tracking branch 'source/0.11.2-rc' into session_persistence

2025-07-07 12:20:44 -04:00 · 2025-07-07 12:20:44 -04:00 · 083331547e
commit 083331547e
--- a/amqtt/plugins/authentication.py
+++ b/amqtt/plugins/authentication.py
@ -58,7 +58,10 @@ class FileAuthPlugin(BaseAuthPlugin):
            return

        try:
-            with Path(password_file).open(mode="r", encoding="utf-8") as file:
+            file = password_file
+            if isinstance(file, str):
+                file = Path(file)
+            with file.open(mode="r", encoding="utf-8") as file:
                self.context.logger.debug(f"Reading user database from {password_file}")
                for _line in file:
                    line = _line.strip()
@ -106,4 +109,4 @@ class FileAuthPlugin(BaseAuthPlugin):
    class Config:
        """Path to the properly encoded password file."""

-        password_file: str | None = None
+        password_file: str | Path | None = None
--- a/amqtt/plugins/manager.py
+++ b/amqtt/plugins/manager.py
@ -84,14 +84,45 @@ class PluginManager(Generic[C]):
        return self.context

    def _load_plugins(self, namespace: str | None = None) -> None:
+        """Load plugins from entrypoint or config dictionary.
+
+        config style is now recommended; entrypoint has been deprecated
+        Example:
+            config = {
+                'listeners':...,
+                'plugins': {
+                    'myproject.myfile.MyPlugin': {}
+            }
+        """
        if self.app_context.config and self.app_context.config.get("plugins", None) is not None:
+            # plugins loaded directly from config dictionary
+
+
            if "auth" in self.app_context.config:
                self.logger.warning("Loading plugins from config will ignore 'auth' section of config")
            if "topic-check" in self.app_context.config:
                self.logger.warning("Loading plugins from config will ignore 'topic-check' section of config")

-            plugin_list: list[Any] = self.app_context.config.get("plugins", [])
-            self._load_str_plugins(plugin_list)
+            plugins_config: list[Any] | dict[str, Any] = self.app_context.config.get("plugins", [])
+
+            # if the config was generated from yaml, the plugins maybe a list instead of a dictionary; transform before loading
+            #
+            # plugins:
+            #   - myproject.myfile.MyPlugin:
+
+            if isinstance(plugins_config, list):
+                plugins_info: dict[str, Any] = {}
+                for plugin_config in plugins_config:
+                    if isinstance(plugin_config, str):
+                        plugins_info.update({plugin_config: {}})
+                    elif not isinstance(plugin_config, dict):
+                        msg = "malformed 'plugins' configuration"
+                        raise PluginLoadError(msg)
+                    else:
+                        plugins_info.update(plugin_config)
+                self._load_str_plugins(plugins_info)
+            elif isinstance(plugins_config, dict):
+                self._load_str_plugins(plugins_config)
        else:
            if not namespace:
                msg = "Namespace needs to be provided for EntryPoint plugin definitions"
@ -106,6 +137,7 @@ class PluginManager(Generic[C]):

            self._load_ep_plugins(namespace)

+        # for all the loaded plugins, find all event callbacks
        for plugin in self._plugins:
            for event in list(BrokerEvents) + list(MQTTEvents):
                if awaitable := getattr(plugin, f"on_{event}", None):
@ -116,7 +148,7 @@ class PluginManager(Generic[C]):
                    self._event_plugin_callbacks[event].append(awaitable)

    def _load_ep_plugins(self, namespace:str) -> None:
-
+        """Load plugins from `pyproject.toml` entrypoints. Deprecated."""
        self.logger.debug(f"Loading plugins for namespace {namespace}")
        auth_filter_list = []
        topic_filter_list = []
@ -146,6 +178,7 @@ class PluginManager(Generic[C]):
                self.logger.debug(f" Plugin {item.name} ready")

    def _load_ep_plugin(self, ep: EntryPoint) -> Plugin | None:
+        """Load plugins from `pyproject.toml` entrypoints. Deprecated."""
        try:
            self.logger.debug(f" Loading plugin {ep!s}")
            plugin = ep.load()
@ -165,40 +198,31 @@ class PluginManager(Generic[C]):
            self.logger.debug(f"Plugin init failed: {ep!r}", exc_info=True)
            raise PluginInitError(ep) from e

-    def _load_str_plugins(self, plugin_list: list[Any]) -> None:
+    def _load_str_plugins(self, plugins_info: dict[str, Any]) -> None:

        self.logger.info("Loading plugins from config")
+        # legacy had a filtering 'enabled' flag, even if plugins were loaded/listed
        self._is_topic_filtering_enabled = True
        self._is_auth_filtering_enabled = True
-        for plugin_info in plugin_list:
-
-            if isinstance(plugin_info, dict):
-                if len(plugin_info.keys()) > 1:
-                    msg = f"config file should have only one key: {plugin_info.keys()}"
-                    raise ValueError(msg)
-                plugin_path = next(iter(plugin_info.keys()))
-                plugin_cfg = plugin_info[plugin_path]
-                plugin = self._load_str_plugin(plugin_path, plugin_cfg)
-            elif isinstance(plugin_info, str):
-                plugin = self._load_str_plugin(plugin_info, {})
-            else:
-                msg = "Unexpected entry in plugins config"
-                raise PluginLoadError(msg)
+        for plugin_path, plugin_config in plugins_info.items():

+            plugin = self._load_str_plugin(plugin_path, plugin_config)
            self._plugins.append(plugin)
+
+            # make sure that authenticate and topic filtering plugins have the appropriate async signature
            if isinstance(plugin, BaseAuthPlugin):
                if not iscoroutinefunction(plugin.authenticate):
-                    msg = f"Auth plugin {plugin_info} has non-async authenticate method."
+                    msg = f"Auth plugin {plugin_path} has non-async authenticate method."
                    raise PluginCoroError(msg)
                self._auth_plugins.append(plugin)
            if isinstance(plugin, BaseTopicPlugin):
                if not iscoroutinefunction(plugin.topic_filtering):
-                    msg = f"Topic plugin {plugin_info} has non-async topic_filtering method."
+                    msg = f"Topic plugin {plugin_path} has non-async topic_filtering method."
                    raise PluginCoroError(msg)
                self._topic_plugins.append(plugin)

    def _load_str_plugin(self, plugin_path: str, plugin_cfg: dict[str, Any] | None = None) -> "BasePlugin[C]":
-
+        """Load plugin from string dotted path: mymodule.myfile.MyPlugin."""
        try:
            plugin_class: Any =  import_string(plugin_path)
        except ImportError as ep:
@ -212,6 +236,8 @@ class PluginManager(Generic[C]):
        plugin_context = copy.copy(self.app_context)
        plugin_context.logger = self.logger.getChild(plugin_class.__name__)
        try:
+            # populate the config based on the inner dataclass called `Config`
+            #   use `dacite` package to type check
            plugin_context.config = from_dict(data_class=plugin_class.Config,
                                              data=plugin_cfg or {},
                                              config=DaciteConfig(strict=True))
@ -232,6 +258,8 @@ class PluginManager(Generic[C]):
    def get_plugin(self, name: str) -> Optional["BasePlugin[C]"]:
        """Get a plugin by its name from the plugins loaded for the current namespace.

+        Only used for testing purposes to verify plugin loading correctly.
+
        :param name:
        :return:
        """
--- a/amqtt/plugins/sys/broker.py
+++ b/amqtt/plugins/sys/broker.py
@ -240,4 +240,4 @@ class BrokerSysPlugin(BasePlugin[BrokerContext]):
    class Config:
        """Configuration struct for plugin."""

-        sys_interval: int = 0
+        sys_interval: int = 20
--- a/amqtt/scripts/default_broker.yaml
+++ b/amqtt/scripts/default_broker.yaml
@ -4,9 +4,9 @@ listeners:
    type: tcp
    bind: 0.0.0.0:1883
 plugins:
-  - amqtt.plugins.logging_amqtt.EventLoggerPlugin:
-  - amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
-  - amqtt.plugins.authentication.AnonymousAuthPlugin:
-      allow_anonymous: true
-  - amqtt.plugins.sys.broker.BrokerSysPlugin:
-      sys_interval: 20
+  amqtt.plugins.logging_amqtt.EventLoggerPlugin:
+  amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
+  amqtt.plugins.authentication.AnonymousAuthPlugin:
+    allow_anonymous: true
+  amqtt.plugins.sys.broker.BrokerSysPlugin:
+    sys_interval: 20
--- a/amqtt/scripts/default_client.yaml
+++ b/amqtt/scripts/default_client.yaml
@ -10,4 +10,4 @@ reconnect_retries: 2
 broker:
  uri: "mqtt://127.0.0.1"
 plugins:
-  - amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
+  amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
--- a/docs/custom_plugins.md
+++ b/docs/custom_plugins.md
@ -40,9 +40,9 @@ dictionary passed to the `Broker` or `MQTTClient`).
 ...
 ...
 plugins:
-  - module.submodule.file.OneClassName:
-  - module.submodule.file.TwoClassName:
-      option1: 123
+  module.submodule.file.OneClassName:
+  module.submodule.file.TwoClassName:
+    option1: 123
 ```

 ??? warning "Deprecated: activating plugins using `EntryPoints`"
--- a/docs/packaged_plugins.md
+++ b/docs/packaged_plugins.md
@ -47,11 +47,10 @@ By default, the `PacketLoggerPlugin` is  activated  and configured for the clien

 ```yaml
 plugins:
-  - ...
-  - amqtt.plugins.authentication.AnonymousAuthPlugin:
-      allow_anonymous: false 
-  - ...
-
+  .
+  .
+  amqtt.plugins.authentication.AnonymousAuthPlugin:
+      allow_anonymous: false
 ```

 !!! danger
@ -78,10 +77,8 @@ clients are authorized by providing username and password, compared against file

 ```yaml
 plugins:
-  - ...
-  - amqtt.plugins.authentication.FileAuthPlugin:
-      password_file: /path/to/password_file 
-  - ...
+  amqtt.plugins.authentication.FileAuthPlugin:
+      password_file: /path/to/password_file
 ```

 ??? warning "EntryPoint-style configuration is deprecated"
@ -119,9 +116,7 @@ Prevents using topics named: `prohibited`, `top-secret`, and `data/classified`

 ```yaml
 plugins:
-  - ...
-  - amqtt.plugins.topic_checking.TopicTabooPlugin:
-  - ...
+  amqtt.plugins.topic_checking.TopicTabooPlugin:
 ```

 ??? warning "EntryPoint-style configuration is deprecated"
@ -139,13 +134,13 @@ plugins:

 **Configuration**

- `acl` *(mapping)*: determines subscription access; if `publish-acl` is not specified, determine both publish and subscription access.
+- `acl` *(mapping)*: determines subscription access
   The list should be a key-value pair, where:
-`<username>:[<topic1>, <topic2>, ...]` *(string, list[string])*: username of the client followed by a list of allowed topics (wildcards are supported: `#`, `+`).
+        `<username>:[<topic1>, <topic2>, ...]` *(string, list[string])*: username of the client followed by a list of allowed topics (wildcards are supported: `#`, `+`).


- `publish-acl` *(mapping)*: determines publish access. This parameter defines the list of access control rules; each item is a key-value pair, where:
-`<username>:[<topic1>, <topic2>, ...]` *(string, list[string])*: username of the client followed by a list of allowed topics (wildcards are supported: `#`, `+`).
+- `publish-acl` *(mapping)*: determines publish access. If absent, no restrictions are placed on client publishing. 
+        `<username>:[<topic1>, <topic2>, ...]` *(string, list[string])*: username of the client followed by a list of allowed topics (wildcards are supported: `#`, `+`).

    !!! info "Reserved usernames"

@ -154,13 +149,13 @@ plugins:

 ```yaml
 plugins:
-  - ...
-  - amqtt.plugins.topic_checking.TopicAccessControlListPlugin:
-      publish_acl:
-        - username: ["list", "of", "allowed", "topics", "for", "publishing"]
-      acl:
-        - username: ["list", "of", "allowed", "topics", "for", "subscribing"]
-  - ...
+  amqtt.plugins.topic_checking.TopicAccessControlListPlugin:
+    acl:
+      - username: ["list", "of", "allowed", "topics", "for", "subscribing"]
+      - .
+    publish_acl:
+      - username: ["list", "of", "allowed", "topics", "for", "publishing"]
+      - .
 ```

 ??? warning "EntryPoint-style configuration is deprecated"
@ -185,13 +180,12 @@ Publishes, on a periodic basis, statistics about the broker

 **Configuration**

- `sys_interval` - int, seconds between updates
+- `sys_interval` - int, seconds between updates (default: 20)
+
 ```yaml
 plugins:
-  - ...
-  - amqtt.plugins.sys.broker.BrokerSysPlugin:
-      sys_interval: 20  # int, seconds between updates 
-  - ...
+  amqtt.plugins.sys.broker.BrokerSysPlugin:
+    sys_interval: 20  # int, seconds between updates
 ```

 **Supported Topics**
@ -231,6 +225,11 @@ This plugin issues log messages when [broker and mqtt events](custom_plugins.md#
 - info level messages for `client connected` and `client disconnected`
 - debug level for all others

+```yaml
+plugins:
+  amqtt.plugins.logging_amqtt.EventLoggerPlugin:
+```
+

 ### Packet Logger

@ -239,3 +238,7 @@ This plugin issues log messages when [broker and mqtt events](custom_plugins.md#
 This plugin issues debug-level messages for [mqtt events](custom_plugins.md#client-and-broker): `on_mqtt_packet_sent`
 and `on_mqtt_packet_received`.

+```yaml
+plugins:
+  amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
+```
--- a/samples/broker_acl.py
+++ b/samples/broker_acl.py
@ -1,6 +1,7 @@
 import asyncio
 import logging
 import os
+from pathlib import Path

 from amqtt.broker import Broker

@ -22,24 +23,20 @@ config = {
            "max_connections": 10,
        },
    },
-    "sys_interval": 10,
-    "auth": {
-        "allow-anonymous": True,
-        "password-file": os.path.join(
-            os.path.dirname(os.path.realpath(__file__)),
-            "passwd",
-        ),
-        "plugins": ["auth_file", "auth_anonymous"],
-    },
-    "topic-check": {
-        "enabled": True,
-        "plugins": ["topic_acl"],
-        "acl": {
-            # username: [list of allowed topics]
-            "test": ["repositories/+/master", "calendar/#", "data/memes"],
-            "anonymous": [],
+    "plugins": {
+        'amqtt.plugins.authentication.AnonymousAuthPlugin': { 'allow_anonymous': True},
+        'amqtt.plugins.authentication.FileAuthPlugin': {
+            'password_file': Path(__file__).parent / 'passwd',
        },
-    },
+        'amqtt.plugins.sys.broker.BrokerSysPlugin': { "sys_interval": 10},
+        'amqtt.plugins.topic_checking.TopicAccessControlListPlugin': {
+            'acl': {
+                # username: [list of allowed topics]
+                "test": ["repositories/+/master", "calendar/#", "data/memes"],
+                "anonymous": [],
+            }
+        }
+    }
 }


--- a/samples/broker_start.py
+++ b/samples/broker_start.py
@ -1,6 +1,7 @@
 import asyncio
 import logging
 import os
+from pathlib import Path

 from amqtt.broker import Broker

@ -22,16 +23,13 @@ config = {
            "max_connections": 10,
        },
    },
-    "sys_interval": 10,
-    "auth": {
-        "allow-anonymous": True,
-        "password-file": os.path.join(
-            os.path.dirname(os.path.realpath(__file__)),
-            "passwd",
-        ),
-        "plugins": ["auth_file", "auth_anonymous"],
-    },
-    "topic-check": {"enabled": False},
+    "plugins": {
+        'amqtt.plugins.authentication.AnonymousAuthPlugin': { 'allow_anonymous': True},
+        'amqtt.plugins.authentication.FileAuthPlugin': {
+            'password_file': Path(__file__).parent / 'passwd',
+        },
+        'amqtt.plugins.sys.broker.BrokerSysPlugin': { "sys_interval": 10},
+    }
 }

 async def main_loop():
--- a/samples/broker_taboo.py
+++ b/samples/broker_taboo.py
@ -1,6 +1,7 @@
 import asyncio
 import logging
 import os
+from pathlib import Path

 from amqtt.broker import Broker

@ -22,16 +23,14 @@ config = {
            "max_connections": 10,
        },
    },
-    "sys_interval": 10,
-    "auth": {
-        "allow-anonymous": True,
-        "password-file": os.path.join(
-            os.path.dirname(os.path.realpath(__file__)),
-            "passwd",
-        ),
-        "plugins": ["auth_file", "auth_anonymous"],
-    },
-    "topic-check": {"enabled": True, "plugins": ["topic_taboo"]},
+    "plugins": {
+        'amqtt.plugins.authentication.AnonymousAuthPlugin': {'allow_anonymous': True},
+        'amqtt.plugins.authentication.FileAuthPlugin': {
+            'password_file': Path(__file__).parent / 'passwd',
+        },
+        'amqtt.plugins.sys.broker.BrokerSysPlugin': {"sys_interval": 10},
+        'amqtt.plugins.topic_checking.TopicTabooPlugin': {},
+    }
 }


--- a/tests/plugins/mocks.py
+++ b/tests/plugins/mocks.py
@ -25,6 +25,7 @@ class TestConfigPlugin(BasePlugin):
    class Config:
        option1: int
        option2: str
+        option3: int = 20


 class TestCoroErrorPlugin(BaseAuthPlugin):
--- a/tests/plugins/test_config.py
+++ b/tests/plugins/test_config.py
@ -219,3 +219,101 @@ async def test_block_topic_plugin_load():
    await client1.disconnect()

    await broker.shutdown()
+
+plugin_yaml_list_config_one = """---
+listeners:
+  default:
+    type: tcp
+    bind: 0.0.0.0:1883
+plugins:
+   - tests.plugins.mocks.TestSimplePlugin:
+   - tests.plugins.mocks.TestConfigPlugin:
+       option1: 1
+       option2: bar
+       option3: 3
+"""
+
+plugin_yaml_list_config_two = """---
+listeners:
+  default:
+    type: tcp
+    bind: 0.0.0.0:1883
+plugins:
+   - tests.plugins.mocks.TestSimplePlugin
+   - tests.plugins.mocks.TestConfigPlugin:
+       option1: 1
+       option2: bar
+       option3: 3
+"""
+
+plugin_yaml_dict_config = """---
+listeners:
+  default:
+    type: tcp
+    bind: 0.0.0.0:1883
+plugins:
+   tests.plugins.mocks.TestSimplePlugin:
+   tests.plugins.mocks.TestConfigPlugin:
+    option1: 1
+    option2: bar
+    option3: 3
+"""
+
+plugin_empty_dict_config = {
+    'listeners': {'default': {'type': 'tcp', 'bind': '127.0.0.1'}},
+    'plugins': {
+        'tests.plugins.mocks.TestSimplePlugin': {},
+    }
+}
+
+plugin_dict_option_config = {
+    'listeners': {'default': {'type': 'tcp', 'bind': '127.0.0.1'}},
+    'plugins': {
+        'tests.plugins.mocks.TestConfigPlugin': {'option1': 1, 'option2': 'bar', 'option3': 3}
+    }
+}
+
+@pytest.mark.asyncio
+async def test_plugin_yaml_list_config():
+    cfg: dict[str, Any] = yaml.load(plugin_yaml_list_config_one, Loader=Loader)
+    broker = Broker(config=cfg)
+
+    await asyncio.sleep(0.5)
+    plugin = broker.plugins_manager.get_plugin('TestConfigPlugin')
+    assert getattr(plugin.context.config, 'option1', None) == 1
+    assert getattr(plugin.context.config, 'option3', None) == 3
+
+    cfg: dict[str, Any] = yaml.load(plugin_yaml_list_config_two, Loader=Loader)
+    broker = Broker(config=cfg)
+
+    await asyncio.sleep(0.5)
+    plugin = broker.plugins_manager.get_plugin('TestConfigPlugin')
+    assert getattr(plugin.context.config, 'option1', None) == 1
+    assert getattr(plugin.context.config, 'option3', None) == 3
+
+
+@pytest.mark.asyncio
+async def test_plugin_yaml_dict_config():
+    cfg: dict[str, Any] = yaml.load(plugin_yaml_dict_config, Loader=Loader)
+    broker = Broker(config=cfg)
+
+    await asyncio.sleep(0.5)
+    assert broker.plugins_manager.get_plugin('TestSimplePlugin') is not None
+
+
+@pytest.mark.asyncio
+async def test_plugin_empty_dict_config():
+    broker = Broker(config=plugin_empty_dict_config)
+
+    await asyncio.sleep(0.5)
+    assert broker.plugins_manager.get_plugin('TestSimplePlugin') is not None
+
+
+@pytest.mark.asyncio
+async def test_plugin_option_dict_config():
+    broker = Broker(config=plugin_dict_option_config)
+
+    await asyncio.sleep(0.5)
+    plugin = broker.plugins_manager.get_plugin('TestConfigPlugin')
+    assert getattr(plugin.context.config, 'option1', None) == 1
+    assert getattr(plugin.context.config, 'option3', None) == 3