kopia lustrzana https://github.com/Yakifo/amqtt
match yaml 'plugins' format as dictionary with python dictionary format; allow for list as well as dictionary, in case that format slips in
rodzic
6f724b9a23
commit
1971af26a4
|
@ -88,8 +88,20 @@ class PluginManager(Generic[C]):
|
||||||
if "topic-check" in self.app_context.config:
|
if "topic-check" in self.app_context.config:
|
||||||
self.logger.warning("Loading plugins from config will ignore 'topic-check' section of 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", [])
|
plugins_config: list[Any] | dict[str, Any] = self.app_context.config.get("plugins", [])
|
||||||
self._load_str_plugins(plugin_list)
|
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)
|
||||||
|
if isinstance(plugins_config, dict):
|
||||||
|
self._load_str_plugins(plugins_config)
|
||||||
else:
|
else:
|
||||||
if not namespace:
|
if not namespace:
|
||||||
msg = "Namespace needs to be provided for EntryPoint plugin definitions"
|
msg = "Namespace needs to be provided for EntryPoint plugin definitions"
|
||||||
|
@ -163,35 +175,24 @@ class PluginManager(Generic[C]):
|
||||||
self.logger.debug(f"Plugin init failed: {ep!r}", exc_info=True)
|
self.logger.debug(f"Plugin init failed: {ep!r}", exc_info=True)
|
||||||
raise PluginInitError(ep) from e
|
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")
|
self.logger.info("Loading plugins from config")
|
||||||
self._is_topic_filtering_enabled = True
|
self._is_topic_filtering_enabled = True
|
||||||
self._is_auth_filtering_enabled = True
|
self._is_auth_filtering_enabled = True
|
||||||
for plugin_info in plugin_list:
|
for plugin_path, plugin_config in plugins_info.items():
|
||||||
|
|
||||||
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)
|
|
||||||
|
|
||||||
|
plugin = self._load_str_plugin(plugin_path, plugin_config)
|
||||||
self._plugins.append(plugin)
|
self._plugins.append(plugin)
|
||||||
|
|
||||||
if isinstance(plugin, BaseAuthPlugin):
|
if isinstance(plugin, BaseAuthPlugin):
|
||||||
if not iscoroutinefunction(plugin.authenticate):
|
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)
|
raise PluginCoroError(msg)
|
||||||
self._auth_plugins.append(plugin)
|
self._auth_plugins.append(plugin)
|
||||||
if isinstance(plugin, BaseTopicPlugin):
|
if isinstance(plugin, BaseTopicPlugin):
|
||||||
if not iscoroutinefunction(plugin.topic_filtering):
|
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)
|
raise PluginCoroError(msg)
|
||||||
self._topic_plugins.append(plugin)
|
self._topic_plugins.append(plugin)
|
||||||
|
|
||||||
|
|
|
@ -4,9 +4,9 @@ listeners:
|
||||||
type: tcp
|
type: tcp
|
||||||
bind: 0.0.0.0:1883
|
bind: 0.0.0.0:1883
|
||||||
plugins:
|
plugins:
|
||||||
- amqtt.plugins.logging_amqtt.EventLoggerPlugin:
|
amqtt.plugins.logging_amqtt.EventLoggerPlugin:
|
||||||
- amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
|
amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
|
||||||
- amqtt.plugins.authentication.AnonymousAuthPlugin:
|
amqtt.plugins.authentication.AnonymousAuthPlugin:
|
||||||
allow_anonymous: true
|
allow_anonymous: true
|
||||||
- amqtt.plugins.sys.broker.BrokerSysPlugin:
|
amqtt.plugins.sys.broker.BrokerSysPlugin:
|
||||||
sys_interval: 20
|
sys_interval: 20
|
|
@ -10,4 +10,4 @@ reconnect_retries: 2
|
||||||
broker:
|
broker:
|
||||||
uri: "mqtt://127.0.0.1"
|
uri: "mqtt://127.0.0.1"
|
||||||
plugins:
|
plugins:
|
||||||
- amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
|
amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
|
||||||
|
|
|
@ -40,9 +40,9 @@ dictionary passed to the `Broker` or `MQTTClient`).
|
||||||
...
|
...
|
||||||
...
|
...
|
||||||
plugins:
|
plugins:
|
||||||
- module.submodule.file.OneClassName:
|
module.submodule.file.OneClassName:
|
||||||
- module.submodule.file.TwoClassName:
|
module.submodule.file.TwoClassName:
|
||||||
option1: 123
|
option1: 123
|
||||||
```
|
```
|
||||||
|
|
||||||
??? warning "Deprecated: activating plugins using `EntryPoints`"
|
??? warning "Deprecated: activating plugins using `EntryPoints`"
|
||||||
|
|
|
@ -47,11 +47,10 @@ By default, the `PacketLoggerPlugin` is activated and configured for the clien
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
plugins:
|
plugins:
|
||||||
- ...
|
.
|
||||||
- amqtt.plugins.authentication.AnonymousAuthPlugin:
|
.
|
||||||
allow_anonymous: false
|
amqtt.plugins.authentication.AnonymousAuthPlugin:
|
||||||
- ...
|
allow_anonymous: false
|
||||||
|
|
||||||
```
|
```
|
||||||
|
|
||||||
!!! danger
|
!!! danger
|
||||||
|
@ -78,10 +77,8 @@ clients are authorized by providing username and password, compared against file
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
plugins:
|
plugins:
|
||||||
- ...
|
amqtt.plugins.authentication.FileAuthPlugin:
|
||||||
- amqtt.plugins.authentication.FileAuthPlugin:
|
password_file: /path/to/password_file
|
||||||
password_file: /path/to/password_file
|
|
||||||
- ...
|
|
||||||
```
|
```
|
||||||
|
|
||||||
??? warning "EntryPoint-style configuration is deprecated"
|
??? warning "EntryPoint-style configuration is deprecated"
|
||||||
|
@ -119,9 +116,7 @@ Prevents using topics named: `prohibited`, `top-secret`, and `data/classified`
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
plugins:
|
plugins:
|
||||||
- ...
|
amqtt.plugins.topic_checking.TopicTabooPlugin:
|
||||||
- amqtt.plugins.topic_checking.TopicTabooPlugin:
|
|
||||||
- ...
|
|
||||||
```
|
```
|
||||||
|
|
||||||
??? warning "EntryPoint-style configuration is deprecated"
|
??? warning "EntryPoint-style configuration is deprecated"
|
||||||
|
@ -139,13 +134,13 @@ plugins:
|
||||||
|
|
||||||
**Configuration**
|
**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:
|
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:
|
- `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: `#`, `+`).
|
`<username>:[<topic1>, <topic2>, ...]` *(string, list[string])*: username of the client followed by a list of allowed topics (wildcards are supported: `#`, `+`).
|
||||||
|
|
||||||
!!! info "Reserved usernames"
|
!!! info "Reserved usernames"
|
||||||
|
|
||||||
|
@ -154,13 +149,13 @@ plugins:
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
plugins:
|
plugins:
|
||||||
- ...
|
amqtt.plugins.topic_checking.TopicAccessControlListPlugin:
|
||||||
- amqtt.plugins.topic_checking.TopicAccessControlListPlugin:
|
acl:
|
||||||
publish_acl:
|
- username: ["list", "of", "allowed", "topics", "for", "subscribing"]
|
||||||
- username: ["list", "of", "allowed", "topics", "for", "publishing"]
|
- .
|
||||||
acl:
|
publish_acl:
|
||||||
- username: ["list", "of", "allowed", "topics", "for", "subscribing"]
|
- username: ["list", "of", "allowed", "topics", "for", "publishing"]
|
||||||
- ...
|
- .
|
||||||
```
|
```
|
||||||
|
|
||||||
??? warning "EntryPoint-style configuration is deprecated"
|
??? warning "EntryPoint-style configuration is deprecated"
|
||||||
|
@ -186,12 +181,11 @@ Publishes, on a periodic basis, statistics about the broker
|
||||||
**Configuration**
|
**Configuration**
|
||||||
|
|
||||||
- `sys_interval` - int, seconds between updates
|
- `sys_interval` - int, seconds between updates
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
plugins:
|
plugins:
|
||||||
- ...
|
amqtt.plugins.sys.broker.BrokerSysPlugin:
|
||||||
- amqtt.plugins.sys.broker.BrokerSysPlugin:
|
sys_interval: 20 # int, seconds between updates
|
||||||
sys_interval: 20 # int, seconds between updates
|
|
||||||
- ...
|
|
||||||
```
|
```
|
||||||
|
|
||||||
**Supported Topics**
|
**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`
|
- info level messages for `client connected` and `client disconnected`
|
||||||
- debug level for all others
|
- debug level for all others
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
plugins:
|
||||||
|
amqtt.plugins.logging_amqtt.EventLoggerPlugin:
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
### Packet Logger
|
### 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`
|
This plugin issues debug-level messages for [mqtt events](custom_plugins.md#client-and-broker): `on_mqtt_packet_sent`
|
||||||
and `on_mqtt_packet_received`.
|
and `on_mqtt_packet_received`.
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
plugins:
|
||||||
|
amqtt.plugins.logging_amqtt.PacketLoggerPlugin:
|
||||||
|
```
|
||||||
|
|
|
@ -25,6 +25,7 @@ class TestConfigPlugin(BasePlugin):
|
||||||
class Config:
|
class Config:
|
||||||
option1: int
|
option1: int
|
||||||
option2: str
|
option2: str
|
||||||
|
option3: int = 20
|
||||||
|
|
||||||
|
|
||||||
class TestCoroErrorPlugin(BaseAuthPlugin):
|
class TestCoroErrorPlugin(BaseAuthPlugin):
|
||||||
|
|
|
@ -219,3 +219,101 @@ async def test_block_topic_plugin_load():
|
||||||
await client1.disconnect()
|
await client1.disconnect()
|
||||||
|
|
||||||
await broker.shutdown()
|
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
|
||||||
|
|
Ładowanie…
Reference in New Issue