From 2c5341a2a5a0ff560d6c2715542454e24c7f04ee Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Fri, 29 Mar 2024 16:16:40 -0400 Subject: [PATCH] Improved variable parsing in YAML files (#1088) --- apprise/URLBase.py | 121 ++++++++++++++--------- apprise/config/ConfigBase.py | 3 + apprise/decorators/CustomNotifyPlugin.py | 4 + test/test_decorator_notify.py | 19 +++- test/test_plugin_email.py | 58 +++++++++++ 5 files changed, 154 insertions(+), 51 deletions(-) diff --git a/apprise/URLBase.py b/apprise/URLBase.py index 2467a4c1..90ea85c6 100644 --- a/apprise/URLBase.py +++ b/apprise/URLBase.py @@ -669,6 +669,79 @@ class URLBase: 'verify': 'yes' if self.verify_certificate else 'no', } + @staticmethod + def post_process_parse_url_results(results): + """ + After parsing the URL, this function applies a bit of extra logic to + support extra entries like `pass` becoming `password`, etc + + This function assumes that parse_url() was called previously setting + up the basics to be checked + """ + + # if our URL ends with an 's', then assume our secure flag is set. + results['secure'] = (results['schema'][-1] == 's') + + # QSD Checking (over-rides all) + qsd_exists = True if isinstance(results.get('qsd'), dict) else False + + if qsd_exists and 'verify' in results['qsd']: + # Pulled from URL String + results['verify'] = parse_bool( + results['qsd'].get('verify', True)) + + elif 'verify' in results: + # Pulled from YAML Configuratoin + results['verify'] = parse_bool(results.get('verify', True)) + + else: + # Support SSL Certificate 'verify' keyword. Default to being + # enabled + results['verify'] = True + + # Password overrides + if 'pass' in results: + results['password'] = results['pass'] + del results['pass'] + + if qsd_exists: + if 'password' in results['qsd']: + results['password'] = results['qsd']['password'] + if 'pass' in results['qsd']: + results['password'] = results['qsd']['pass'] + + # User overrides + if 'user' in results['qsd']: + results['user'] = results['qsd']['user'] + + # parse_url() always creates a 'password' and 'user' entry in the + # results returned. Entries are set to None if they weren't + # specified + if results['password'] is None and 'user' in results['qsd']: + # Handle cases where the user= provided in 2 locations, we want + # the original to fall back as a being a password (if one + # wasn't otherwise defined) e.g. + # mailtos://PASSWORD@hostname?user=admin@mail-domain.com + # - in the above, the PASSWORD gets lost in the parse url() + # since a user= over-ride is specified. + presults = parse_url(results['url']) + if presults: + # Store our Password + results['password'] = presults['user'] + + # Store our socket read timeout if specified + if 'rto' in results['qsd']: + results['rto'] = results['qsd']['rto'] + + # Store our socket connect timeout if specified + if 'cto' in results['qsd']: + results['cto'] = results['qsd']['cto'] + + if 'port' in results['qsd']: + results['port'] = results['qsd']['port'] + + return results + @staticmethod def parse_url(url, verify_host=True, plus_to_space=False, strict_port=False): @@ -698,53 +771,7 @@ class URLBase: # We're done; we failed to parse our url return results - # if our URL ends with an 's', then assume our secure flag is set. - results['secure'] = (results['schema'][-1] == 's') - - # Support SSL Certificate 'verify' keyword. Default to being enabled - results['verify'] = True - - if 'verify' in results['qsd']: - results['verify'] = parse_bool( - results['qsd'].get('verify', True)) - - # Password overrides - if 'password' in results['qsd']: - results['password'] = results['qsd']['password'] - if 'pass' in results['qsd']: - results['password'] = results['qsd']['pass'] - - # User overrides - if 'user' in results['qsd']: - results['user'] = results['qsd']['user'] - - # parse_url() always creates a 'password' and 'user' entry in the - # results returned. Entries are set to None if they weren't specified - if results['password'] is None and 'user' in results['qsd']: - # Handle cases where the user= provided in 2 locations, we want - # the original to fall back as a being a password (if one wasn't - # otherwise defined) - # e.g. - # mailtos://PASSWORD@hostname?user=admin@mail-domain.com - # - the PASSWORD gets lost in the parse url() since a user= - # over-ride is specified. - presults = parse_url(results['url']) - if presults: - # Store our Password - results['password'] = presults['user'] - - # Store our socket read timeout if specified - if 'rto' in results['qsd']: - results['rto'] = results['qsd']['rto'] - - # Store our socket connect timeout if specified - if 'cto' in results['qsd']: - results['cto'] = results['qsd']['cto'] - - if 'port' in results['qsd']: - results['port'] = results['qsd']['port'] - - return results + return URLBase.post_process_parse_url_results(results) @staticmethod def http_response_code_lookup(code, response_mask=None): diff --git a/apprise/config/ConfigBase.py b/apprise/config/ConfigBase.py index 73194525..32e1bde3 100644 --- a/apprise/config/ConfigBase.py +++ b/apprise/config/ConfigBase.py @@ -1184,6 +1184,9 @@ class ConfigBase(URLBase): # Prepare our Asset Object _results['asset'] = asset + # Handle post processing of result set + _results = URLBase.post_process_parse_url_results(_results) + # Store our preloaded entries preloaded.append({ 'results': _results, diff --git a/apprise/decorators/CustomNotifyPlugin.py b/apprise/decorators/CustomNotifyPlugin.py index 4c93ef1b..eb5f17b7 100644 --- a/apprise/decorators/CustomNotifyPlugin.py +++ b/apprise/decorators/CustomNotifyPlugin.py @@ -147,6 +147,10 @@ class CustomNotifyPlugin(NotifyBase): self._default_args = {} + # Some variables do not need to be set + if 'secure' in kwargs: + del kwargs['secure'] + # Apply our updates based on what was parsed dict_full_update(self._default_args, self._base_args) dict_full_update(self._default_args, kwargs) diff --git a/test/test_decorator_notify.py b/test/test_decorator_notify.py index 9d188d35..96baeef5 100644 --- a/test/test_decorator_notify.py +++ b/test/test_decorator_notify.py @@ -29,6 +29,7 @@ from os.path import dirname from os.path import join from apprise.decorators import notify +from apprise.decorators.CustomNotifyPlugin import CustomNotifyPlugin from apprise import Apprise from apprise import AppriseConfig from apprise import AppriseAsset @@ -351,7 +352,7 @@ def test_notify_multi_instance_decoration(tmpdir): t = tmpdir.mkdir("multi-test").join("apprise.yml") t.write("""urls: - multi://user1:pass@hostname - - multi://user2:pass2@hostname + - multi://user2:pass2@hostname?verify=no """) # Create ourselves a config object @@ -404,11 +405,12 @@ def test_notify_multi_instance_decoration(tmpdir): assert 'tag' in meta assert isinstance(meta['tag'], set) - assert len(meta) == 7 + assert len(meta) == 8 # We carry all of our default arguments from the @notify's initialization assert meta['schema'] == 'multi' assert meta['host'] == 'hostname' assert meta['user'] == 'user1' + assert meta['verify'] is True assert meta['password'] == 'pass' # Verify our URL is correct @@ -441,15 +443,24 @@ def test_notify_multi_instance_decoration(tmpdir): assert 'tag' in meta assert isinstance(meta['tag'], set) - assert len(meta) == 7 + assert len(meta) == 9 # We carry all of our default arguments from the @notify's initialization assert meta['schema'] == 'multi' assert meta['host'] == 'hostname' assert meta['user'] == 'user2' assert meta['password'] == 'pass2' + assert meta['verify'] is False + assert meta['qsd']['verify'] == 'no' # Verify our URL is correct - assert meta['url'] == 'multi://user2:pass2@hostname' + assert meta['url'] == 'multi://user2:pass2@hostname?verify=no' # Tidy N_MGR.remove('multi') + + +def test_custom_notify_plugin_decoration(): + """decorators: CustomNotifyPlugin testing + """ + + CustomNotifyPlugin() diff --git a/test/test_plugin_email.py b/test/test_plugin_email.py index 62efc4b9..f1ddd87e 100644 --- a/test/test_plugin_email.py +++ b/test/test_plugin_email.py @@ -30,6 +30,7 @@ import logging import os import re from unittest import mock +from inspect import cleandoc import smtplib from email.header import decode_header @@ -37,6 +38,8 @@ from email.header import decode_header from apprise import NotifyType, NotifyBase from apprise import Apprise from apprise import AttachBase +from apprise.AppriseAsset import AppriseAsset +from apprise.config.ConfigBase import ConfigBase from apprise import AppriseAttachment from apprise.plugins.NotifyEmail import NotifyEmail from apprise.plugins import NotifyEmail as NotifyEmailModule @@ -1757,3 +1760,58 @@ def test_plugin_email_formatting_990(mock_smtp, mock_smtp_ssl): assert len(obj.targets) == 1 assert (False, 'me@mydomain.com') in obj.targets + + +def test_plugin_email_variables_1087(): + """ + NotifyEmail() GitHub Issue 1087 + https://github.com/caronc/apprise/issues/1087 + Email variables reported not working correctly + + """ + + # Valid Configuration + result, _ = ConfigBase.config_parse(cleandoc(""" + # + # Test Email Parsing + # + urls: + - mailtos://alt.lan/: + - user: testuser@alt.lan + pass: xxxxXXXxxx + smtp: smtp.alt.lan + to: alteriks@alt.lan + """), asset=AppriseAsset()) + + assert isinstance(result, list) + assert len(result) == 1 + + email = result[0] + assert email.from_addr == ['Apprise', 'testuser@alt.lan'] + assert email.user == 'testuser@alt.lan' + assert email.smtp_host == 'smtp.alt.lan' + assert email.targets == [(False, 'alteriks@alt.lan')] + assert email.password == 'xxxxXXXxxx' + + # Valid Configuration + result, _ = ConfigBase.config_parse(cleandoc(""" + # + # Test Email Parsing where qsd over-rides all + # + urls: + - mailtos://alt.lan/?pass=abcd&user=joe@alt.lan: + - user: testuser@alt.lan + pass: xxxxXXXxxx + smtp: smtp.alt.lan + to: alteriks@alt.lan + """), asset=AppriseAsset()) + + assert isinstance(result, list) + assert len(result) == 1 + + email = result[0] + assert email.from_addr == ['Apprise', 'joe@alt.lan'] + assert email.user == 'joe@alt.lan' + assert email.smtp_host == 'smtp.alt.lan' + assert email.targets == [(False, 'alteriks@alt.lan')] + assert email.password == 'abcd'