Add configuration variable expansion

docs/configure/mqttwarn.ini.md

@@ -218,3 +218,29 @@ display them on all XBMC targets:
 targets  = log:error, xbmc
 title    = mqttwarn
 ```
+
+## Variables
+
+You can load option values either from environment variables or file content.
+To do this, replace option's value with one of the following:
+
+- `${ENV:FOO}` - Replaces option's value with environment variable `FOO`.
+- `${FILE:/path/to/foo.txt}` - Replaces option's value with file contents from
+  `/path/to/foo.txt`. The file path can also be relative like `${FILE:foo.txt}`
+  in which case the file is loaded relative to configuration file's location.
+
+The variable pattern can take either form like `$TYPE:NAME` or `${TYPE:NAME}`.
+Latter pattern is required when variable name (`NAME`) contains characters that
+are not alphanumeric or underscore.
+
+For example:
+```ini
+[defaults]
+username = $ENV:MQTTWARN_USERNAME
+password = $ENV:MQTTWARN_PASSWORD
+
+[config:xxx]
+targets = {
+    'targetname1': [ '${FILE:/run/secrets/address.txt}' ],
+  }
+```

mqttwarn/configuration.py

@@ -4,9 +4,10 @@
 import codecs
 import logging
 import os
+import re
 import sys
 import typing as t
-from configparser import NoOptionError, NoSectionError, RawConfigParser
+from configparser import Interpolation, NoOptionError, NoSectionError, RawConfigParser
 
 from mqttwarn.util import load_functions
 
@@ -20,6 +21,72 @@
 logger = logging.getLogger(__name__)
 
 
+def expand_vars(input: str, sources: t.Dict[str, t.Callable[[str], str]]) -> str:
+    """
+    Expand variables in `input` string with values from `sources` dict.
+
+    Variables may be in two forms, either $TYPE:KEY or ${TYPE:KEY}. The second form must be used when `KEY` contains
+    characters other than numbers, alphabets or underscore. Supported `TYPE`s depends on keys of `sources` dict.
+
+    The `sources` is a dict where key is name of `TYPE` in the pattern above and value is a function that takes `KEY`
+    as argument and returns contents of the variable to be expanded.
+
+    :return: Input string with variables expanded
+    """
+    expanded = ""
+    input_index = 0
+    match = None
+    # `input` may have multiple variables in form of $TYPE:KEY or ${TYPE:KEY} pattern, iterate through them
+    for match in re.finditer(r"\$(\w+):(\w+)|\$\{(\w+):([^}]+)\}", input):
+        var_type = match[1] if match[1] else match[3]  # TYPE part in the variable pattern
+        var_key = match[2] if match[2] else match[4]  # KEY part in the variable pattern
+
+        if var_type not in sources:
+            raise KeyError(f"{match[0]}: Variable type '{var_type}' not supported")
+        source = sources[var_type]
+
+        try:
+            value = source(var_key)
+        except Exception as ex:
+            raise KeyError(f"{match[0]}: {str(ex)}") from ex
+
+        match_start, match_end = match.span()
+        expanded += input[input_index:match_start] + value
+        input_index = match_end
+
+    if match:
+        return expanded + input[input_index:]
+    return input
+
+
+class VariableInterpolation(Interpolation):
+    def __init__(self, configuration_path):
+        self.configuration_path = configuration_path
+        self.sources = {
+            "ENV": self.get_env_variable,
+            "FILE": self.get_file_contents,
+        }
+
+    def before_get(self, parser, section, option, value, defaults):
+        return expand_vars(value, self.sources) if type(value) == str else value
+
+    def get_env_variable(self, name: str) -> str:
+        """
+        Get environment variable of `name` and return it
+        """
+        return os.environ[name]
+
+    def get_file_contents(self, filepath: str) -> str:
+        """
+        Get file contents from `filepath` and return it
+        """
+        if not os.path.isfile(filepath):
+            # Read file contents relative to path of configuration file if path is relative
+            filepath = os.path.join(self.configuration_path, filepath)
+        with open(filepath) as file:
+            return file.read()
+
+
 class Config(RawConfigParser):
 
     specials: t.Dict[str, t.Union[bool, None]] = {
@@ -34,13 +101,14 @@ def __init__(self, configuration_file: t.Optional[str] = None, defaults: t.Optio
 
         self.configuration_path = None
 
-        RawConfigParser.__init__(self)
+        configuration_path = os.path.dirname(configuration_file) if configuration_file else None
+        RawConfigParser.__init__(self, interpolation=VariableInterpolation(configuration_path))
         if configuration_file is not None:
             f = codecs.open(configuration_file, "r", encoding="utf-8")
             self.read_file(f)
             f.close()
 
-            self.configuration_path = os.path.dirname(configuration_file)
+            self.configuration_path = configuration_path
 
         """ set defaults """
         self.hostname = "localhost"

tests/__init__.py

@@ -8,5 +8,6 @@
 configfile_empty_functions = "tests/etc/empty-functions.ini"
 configfile_logging_levels = "tests/etc/logging-levels.ini"
 configfile_better_addresses = "tests/etc/better-addresses.ini"
+configfile_with_variables = "tests/etc/with-variables.ini"
 funcfile_good = "tests/etc/functions_good.py"
 funcfile_bad = "tests/etc/functions_bad.py"

tests/etc/password.txt

@@ -0,0 +1 @@
+secret-password

tests/etc/with-variables.ini

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+# (c) 2023 The mqttwarn developers
+#
+# mqttwarn configuration file for testing variable expansion.
+#
+
+; -------
+; General
+; -------
+
+[defaults]
+hostname     = $ENV:HOSTNAME
+port         = $ENV:PORT
+username     = ${ENV:USERNAME}
+password     = ${FILE:./password.txt}
+
+; name the service providers you will be using.
+launch    = file
+
+
+; --------
+; Services
+; --------
+
+[config:file]
+targets = {
+    'mylog'   : [ '$ENV:LOG_FILE' ],
+  }

tests/test_configuration.py

@@ -1,13 +1,15 @@
 # -*- coding: utf-8 -*-
 # (c) 2022 The mqttwarn developers
+import os
+import re
 import ssl
-from unittest.mock import Mock, call
+from unittest.mock import Mock, call, patch
 
 import pytest
 
 import mqttwarn.configuration
 from mqttwarn.configuration import load_configuration
-from tests import configfile_better_addresses
+from tests import configfile_better_addresses, configfile_with_variables
 
 
 def test_config_with_ssl():
@@ -81,3 +83,85 @@ def test_config_better_addresses_pushsafer():
     assert apprise_service_targets["nagios"]["device"] == "52|65|78"
     assert apprise_service_targets["nagios"]["priority"] == 2
     assert apprise_service_targets["tracking"]["device"] == "gs23"
+
+
+@patch.dict(os.environ, {"HOSTNAME": "example.com", "PORT": "3000", "USERNAME": "bob", "LOG_FILE": "/tmp/out.log"})
+def test_config_expand_variables():
+    """
+    Verify reading configuration file expands variables.
+    """
+    config = load_configuration(configfile_with_variables)
+    assert config.hostname == "example.com"
+    assert config.port == 3000
+    assert config.username == "bob"
+    assert config.password == "secret-password"
+    assert config.getdict("config:file", "targets")["mylog"][0] == "/tmp/out.log"
+
+
+@pytest.mark.parametrize(
+    "input, expected",
+    [
+        ("my-password", "my-password"),
+        ("$SRC_1:PASSWORD_1", "my-password"),
+        ("$SRC_1:PASSWORD_2", "super-secret"),
+        ("-->$SRC_1:PASSWORD_1<--", "-->my-password<--"),
+        ("$SRC_2:PASSWORD_1", "p4ssw0rd"),
+        ("$SRC_1:PÄSSWÖRD_3", "non-ascii-secret"),
+        ("${SRC_1:PASSWORD_1}", "my-password"),
+        ("${SRC_1:/path/to/password.txt}", "file-contents"),
+        ("${SRC_1:PASSWORD_1} ${SRC_1:PASSWORD_2}", "my-password super-secret"),
+        ("$SRC_1:PASSWORD_1 ${SRC_1:/path/to/password.txt} $SRC_1:PASSWORD_1", "my-password file-contents my-password"),
+        (
+            "${SRC_1:/path/to/password.txt} $SRC_1:PASSWORD_1 ${SRC_1:/path/to/password.txt}",
+            "file-contents my-password file-contents",
+        ),
+        ("/$SRC_1:PASSWORD_1/$SRC_1:PASSWORD_2/foo.txt", "/my-password/super-secret/foo.txt"),
+    ],
+)
+def test_expand_vars_ok(input, expected):
+    """
+    Verify that `expand_vars` expands variables in configuration.
+    """
+
+    def create_source(variables):
+        return lambda name: variables[name]
+
+    sources = {
+        "SRC_1": create_source(
+            {
+                "PASSWORD_1": "my-password",
+                "PASSWORD_2": "super-secret",
+                "PÄSSWÖRD_3": "non-ascii-secret",
+                "/path/to/password.txt": "file-contents",
+            }
+        ),
+        "SRC_2": create_source(
+            {
+                "PASSWORD_1": "p4ssw0rd",
+            }
+        ),
+    }
+    expanded = mqttwarn.configuration.expand_vars(input, sources)
+    assert expanded == expected
+
+
+def test_expand_vars_variable_type_not_supported():
+    """
+    Verify that `expand_vars` raises error when variable type is not supported.
+    """
+    with pytest.raises(
+        KeyError, match=re.escape("$DOES_NOT_EXIST:VARIABLE: Variable type 'DOES_NOT_EXIST' not supported")
+    ):
+        mqttwarn.configuration.expand_vars("-->$DOES_NOT_EXIST:VARIABLE<--", {})
+
+
+def test_expand_vars_variable_not_found():
+    """
+    Verify that `expand_vars` raises error when variable is not in source.
+    """
+
+    def empty_source(name):
+        raise KeyError("Variable not found")
+
+    with pytest.raises(KeyError, match=re.escape("$SRC_1:VARIABLE: 'Variable not found'")):
+        mqttwarn.configuration.expand_vars("-->$SRC_1:VARIABLE<--", {"SRC_1": empty_source})