From 8fa146685f5dd0b729f013882f847ffe58bd576f Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 29 Jan 2022 14:23:44 -0500 Subject: [PATCH] Fullpath bulletproofing on XML, JSON, and FORM plugins (#519) --- apprise/plugins/NotifyForm.py | 18 ++-- apprise/plugins/NotifyJSON.py | 9 +- apprise/plugins/NotifyXML.py | 9 +- test/test_plugin_custom_form.py | 152 ++++++++++++++++++++++++++++++++ test/test_plugin_custom_json.py | 62 +++++++++++++ test/test_plugin_custom_xml.py | 55 ++++++++++++ 6 files changed, 289 insertions(+), 16 deletions(-) diff --git a/apprise/plugins/NotifyForm.py b/apprise/plugins/NotifyForm.py index 43463016..32ae2c06 100644 --- a/apprise/plugins/NotifyForm.py +++ b/apprise/plugins/NotifyForm.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- # -# Copyright (C) 2019 Chris Caron +# Copyright (C) 2022 Chris Caron # All rights reserved. # # This code is licensed under the MIT License. @@ -138,7 +138,7 @@ class NotifyForm(NotifyBase): self.fullpath = kwargs.get('fullpath') if not isinstance(self.fullpath, six.string_types): - self.fullpath = '/' + self.fullpath = '' self.method = self.template_args['method']['default'] \ if not isinstance(method, six.string_types) else method.upper() @@ -195,14 +195,15 @@ class NotifyForm(NotifyBase): default_port = 443 if self.secure else 80 - return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format( + return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format( schema=self.secure_protocol if self.secure else self.protocol, auth=auth, # never encode hostname since we're expecting it to be a valid one hostname=self.host, port='' if self.port is None or self.port == default_port else ':{}'.format(self.port), - fullpath=NotifyForm.quote(self.fullpath, safe='/'), + fullpath=NotifyForm.quote(self.fullpath, safe='/') + if self.fullpath else '/', params=NotifyForm.urlencode(params), ) @@ -279,8 +280,8 @@ class NotifyForm(NotifyBase): url += self.fullpath - self.logger.debug('Form POST URL: %s (cert_verify=%r)' % ( - url, self.verify_certificate, + self.logger.debug('Form %s URL: %s (cert_verify=%r)' % ( + self.method, url, self.verify_certificate, )) self.logger.debug('Form Payload: %s' % str(payload)) @@ -306,13 +307,14 @@ class NotifyForm(NotifyBase): r = method( url, files=None if not files else files, - data=payload, + data=payload if self.method != 'GET' else None, + params=payload if self.method == 'GET' else None, headers=headers, auth=auth, verify=self.verify_certificate, timeout=self.request_timeout, ) - if r.status_code != requests.codes.ok: + if r.status_code < 200 or r.status_code >= 300: # We had a problem status_str = \ NotifyForm.http_response_code_lookup(r.status_code) diff --git a/apprise/plugins/NotifyJSON.py b/apprise/plugins/NotifyJSON.py index 99adfd9c..bdd5362b 100644 --- a/apprise/plugins/NotifyJSON.py +++ b/apprise/plugins/NotifyJSON.py @@ -136,7 +136,7 @@ class NotifyJSON(NotifyBase): self.fullpath = kwargs.get('fullpath') if not isinstance(self.fullpath, six.string_types): - self.fullpath = '/' + self.fullpath = '' self.method = self.template_args['method']['default'] \ if not isinstance(method, six.string_types) else method.upper() @@ -184,14 +184,15 @@ class NotifyJSON(NotifyBase): default_port = 443 if self.secure else 80 - return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format( + return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format( schema=self.secure_protocol if self.secure else self.protocol, auth=auth, # never encode hostname since we're expecting it to be a valid one hostname=self.host, port='' if self.port is None or self.port == default_port else ':{}'.format(self.port), - fullpath=NotifyJSON.quote(self.fullpath, safe='/'), + fullpath=NotifyJSON.quote(self.fullpath, safe='/') + if self.fullpath else '/', params=NotifyJSON.urlencode(params), ) @@ -296,7 +297,7 @@ class NotifyJSON(NotifyBase): verify=self.verify_certificate, timeout=self.request_timeout, ) - if r.status_code != requests.codes.ok: + if r.status_code < 200 or r.status_code >= 300: # We had a problem status_str = \ NotifyJSON.http_response_code_lookup(r.status_code) diff --git a/apprise/plugins/NotifyXML.py b/apprise/plugins/NotifyXML.py index f963b66b..2d36d110 100644 --- a/apprise/plugins/NotifyXML.py +++ b/apprise/plugins/NotifyXML.py @@ -157,7 +157,7 @@ class NotifyXML(NotifyBase): self.fullpath = kwargs.get('fullpath') if not isinstance(self.fullpath, six.string_types): - self.fullpath = '/' + self.fullpath = '' self.method = self.template_args['method']['default'] \ if not isinstance(method, six.string_types) else method.upper() @@ -205,14 +205,15 @@ class NotifyXML(NotifyBase): default_port = 443 if self.secure else 80 - return '{schema}://{auth}{hostname}{port}{fullpath}/?{params}'.format( + return '{schema}://{auth}{hostname}{port}{fullpath}?{params}'.format( schema=self.secure_protocol if self.secure else self.protocol, auth=auth, # never encode hostname since we're expecting it to be a valid one hostname=self.host, port='' if self.port is None or self.port == default_port else ':{}'.format(self.port), - fullpath=NotifyXML.quote(self.fullpath, safe='/'), + fullpath=NotifyXML.quote(self.fullpath, safe='/') + if self.fullpath else '/', params=NotifyXML.urlencode(params), ) @@ -335,7 +336,7 @@ class NotifyXML(NotifyBase): verify=self.verify_certificate, timeout=self.request_timeout, ) - if r.status_code != requests.codes.ok: + if r.status_code < 200 or r.status_code >= 300: # We had a problem status_str = \ NotifyXML.http_response_code_lookup(r.status_code) diff --git a/test/test_plugin_custom_form.py b/test/test_plugin_custom_form.py index 90a4a160..520ee2a1 100644 --- a/test/test_plugin_custom_form.py +++ b/test/test_plugin_custom_form.py @@ -224,3 +224,155 @@ def test_plugin_custom_form_attachments(mock_post): assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is False + + +@mock.patch('requests.post') +@mock.patch('requests.get') +def test_plugin_custom_form_edge_cases(mock_get, mock_post): + """ + NotifyForm() Edge Cases + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + mock_get.return_value = response + + results = plugins.NotifyForm.parse_url( + 'form://localhost:8080/command?:abcd=test&method=POST') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] == 8080 + assert results['host'] == 'localhost' + assert results['fullpath'] == '/command' + assert results['path'] == '/' + assert results['query'] == 'command' + assert results['schema'] == 'form' + assert results['url'] == 'form://localhost:8080/command' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd:']['abcd'] == 'test' + + instance = plugins.NotifyForm(**results) + assert isinstance(instance, plugins.NotifyForm) + + response = instance.send(title='title', body='body') + assert response is True + assert mock_post.call_count == 1 + assert mock_get.call_count == 0 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'http://localhost:8080/command' + assert 'abcd' in details[1]['data'] + assert details[1]['data']['abcd'] == 'test' + assert 'title' in details[1]['data'] + assert details[1]['data']['title'] == 'title' + assert 'message' in details[1]['data'] + assert details[1]['data']['message'] == 'body' + + assert instance.url(privacy=False).startswith( + 'form://localhost:8080/command?') + + # Generate a new URL based on our last and verify key values are the same + new_results = plugins.NotifyForm.parse_url(instance.url(safe=False)) + for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query', + 'schema', 'url', 'payload', 'method'): + assert new_results[k] == results[k] + + # Reset our mock configuration + mock_post.reset_mock() + mock_get.reset_mock() + + results = plugins.NotifyForm.parse_url( + 'form://localhost:8080/command?:message=test&method=POST') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] == 8080 + assert results['host'] == 'localhost' + assert results['fullpath'] == '/command' + assert results['path'] == '/' + assert results['query'] == 'command' + assert results['schema'] == 'form' + assert results['url'] == 'form://localhost:8080/command' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd:']['message'] == 'test' + + instance = plugins.NotifyForm(**results) + assert isinstance(instance, plugins.NotifyForm) + + response = instance.send(title='title', body='body') + assert response is True + assert mock_post.call_count == 1 + assert mock_get.call_count == 0 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'http://localhost:8080/command' + assert 'title' in details[1]['data'] + assert details[1]['data']['title'] == 'title' + # 'body' is over-ridden by 'test' passed inline with the URL + assert 'message' in details[1]['data'] + assert details[1]['data']['message'] == 'test' + + assert instance.url(privacy=False).startswith( + 'form://localhost:8080/command?') + + # Generate a new URL based on our last and verify key values are the same + new_results = plugins.NotifyForm.parse_url(instance.url(safe=False)) + for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query', + 'schema', 'url', 'payload', 'method'): + assert new_results[k] == results[k] + + # Reset our mock configuration + mock_post.reset_mock() + mock_get.reset_mock() + + results = plugins.NotifyForm.parse_url( + 'form://localhost:8080/command?:message=test&method=GET') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] == 8080 + assert results['host'] == 'localhost' + assert results['fullpath'] == '/command' + assert results['path'] == '/' + assert results['query'] == 'command' + assert results['schema'] == 'form' + assert results['url'] == 'form://localhost:8080/command' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd:']['message'] == 'test' + + instance = plugins.NotifyForm(**results) + assert isinstance(instance, plugins.NotifyForm) + + response = instance.send(title='title', body='body') + assert response is True + assert mock_post.call_count == 0 + assert mock_get.call_count == 1 + + details = mock_get.call_args_list[0] + assert details[0][0] == 'http://localhost:8080/command' + + assert 'title' in details[1]['params'] + assert details[1]['params']['title'] == 'title' + # 'body' is over-ridden by 'test' passed inline with the URL + assert 'message' in details[1]['params'] + assert details[1]['params']['message'] == 'test' + + assert instance.url(privacy=False).startswith( + 'form://localhost:8080/command?') + + # Generate a new URL based on our last and verify key values are the same + new_results = plugins.NotifyForm.parse_url(instance.url(safe=False)) + for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query', + 'schema', 'url', 'payload', 'method'): + assert new_results[k] == results[k] diff --git a/test/test_plugin_custom_json.py b/test/test_plugin_custom_json.py index c9824285..7d2569f9 100644 --- a/test/test_plugin_custom_json.py +++ b/test/test_plugin_custom_json.py @@ -22,6 +22,8 @@ # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. +import json +import mock import requests from apprise import plugins from helpers import AppriseURLTester @@ -130,3 +132,63 @@ def test_plugin_custom_json_urls(): # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +@mock.patch('requests.get') +def test_plugin_custom_json_edge_cases(mock_get, mock_post): + """ + NotifyJSON() Edge Cases + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + mock_get.return_value = response + + results = plugins.NotifyJSON.parse_url( + 'json://localhost:8080/command?:message=test&method=GET') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] == 8080 + assert results['host'] == 'localhost' + assert results['fullpath'] == '/command' + assert results['path'] == '/' + assert results['query'] == 'command' + assert results['schema'] == 'json' + assert results['url'] == 'json://localhost:8080/command' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd:']['message'] == 'test' + + instance = plugins.NotifyJSON(**results) + assert isinstance(instance, plugins.NotifyJSON) + + response = instance.send(title='title', body='body') + assert response is True + assert mock_post.call_count == 0 + assert mock_get.call_count == 1 + + details = mock_get.call_args_list[0] + assert details[0][0] == 'http://localhost:8080/command' + assert 'title' in details[1]['data'] + dataset = json.loads(details[1]['data']) + assert dataset['title'] == 'title' + assert 'message' in dataset + assert dataset['message'] == 'body' + + assert instance.url(privacy=False).startswith( + 'json://localhost:8080/command?') + + # Generate a new URL based on our last and verify key values are the same + new_results = plugins.NotifyJSON.parse_url(instance.url(safe=False)) + for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query', + 'schema', 'url', 'method'): + assert new_results[k] == results[k] diff --git a/test/test_plugin_custom_xml.py b/test/test_plugin_custom_xml.py index 86f8914f..644fa0f5 100644 --- a/test/test_plugin_custom_xml.py +++ b/test/test_plugin_custom_xml.py @@ -22,6 +22,7 @@ # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. +import mock import requests from apprise import plugins from helpers import AppriseURLTester @@ -146,3 +147,57 @@ def test_plugin_custom_xml_urls(): # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +@mock.patch('requests.get') +def test_plugin_custom_xml_edge_cases(mock_get, mock_post): + """ + NotifyXML() Edge Cases + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + mock_get.return_value = response + + results = plugins.NotifyXML.parse_url( + 'xml://localhost:8080/command?:message=test&method=GET') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] == 8080 + assert results['host'] == 'localhost' + assert results['fullpath'] == '/command' + assert results['path'] == '/' + assert results['query'] == 'command' + assert results['schema'] == 'xml' + assert results['url'] == 'xml://localhost:8080/command' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd:']['message'] == 'test' + + instance = plugins.NotifyXML(**results) + assert isinstance(instance, plugins.NotifyXML) + + response = instance.send(title='title', body='body') + assert response is True + assert mock_post.call_count == 0 + assert mock_get.call_count == 1 + + details = mock_get.call_args_list[0] + assert details[0][0] == 'http://localhost:8080/command' + assert instance.url(privacy=False).startswith( + 'xml://localhost:8080/command?') + + # Generate a new URL based on our last and verify key values are the same + new_results = plugins.NotifyXML.parse_url(instance.url(safe=False)) + for k in ('user', 'password', 'port', 'host', 'fullpath', 'path', 'query', + 'schema', 'url', 'method'): + assert new_results[k] == results[k]