diff --git a/KEYWORDS b/KEYWORDS index 26331e5c..2a64868a 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -88,6 +88,7 @@ Stride Syslog Techulus Telegram +Threema Gateway Twilio Twist Twitter diff --git a/README.md b/README.md index 5764a7da..b05b3b3b 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,7 @@ The table below identifies the services this tool supports and some example serv | [Signal API](https://github.com/caronc/apprise/wiki/Notify_signal) | signal:// or signals:// | (TCP) 80 or 443 | signal://hostname:port/FromPhoneNo
signal://hostname:port/FromPhoneNo/ToPhoneNo
signal://hostname:port/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Sinch](https://github.com/caronc/apprise/wiki/Notify_sinch) | sinch:// | (TCP) 443 | sinch://ServicePlanId:ApiToken@FromPhoneNo
sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo
sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo
sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [SMSEagle](https://github.com/caronc/apprise/wiki/Notify_smseagle) | smseagle:// or smseagles:// | (TCP) 80 or 443 | smseagles://hostname:port/ToPhoneNo
smseagles://hostname:port/@ToContact
smseagles://hostname:port/#ToGroup
smseagles://hostname:port/ToPhoneNo1/#ToGroup/@ToContact/ +| [Threema Gateway](https://github.com/caronc/apprise/wiki/Notify_threema) | threema:// | (TCP) 443 | threema://GatewayID@secret/ToPhoneNo
threema://GatewayID@secret/ToEmail
threema://GatewayID@secret/ToThreemaID/
threema://GatewayID@secret/ToEmail/ToThreemaID/ToPhoneNo/... | [Twilio](https://github.com/caronc/apprise/wiki/Notify_twilio) | twilio:// | (TCP) 443 | twilio://AccountSid:AuthToken@FromPhoneNo
twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo
twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo?apikey=Key
twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo
twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Voipms](https://github.com/caronc/apprise/wiki/Notify_voipms) | voipms:// | (TCP) 443 | voipms://password:email/FromPhoneNo
voipms://password:email/FromPhoneNo/ToPhoneNo
voipms://password:email/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Vonage](https://github.com/caronc/apprise/wiki/Notify_nexmo) (formerly Nexmo) | nexmo:// | (TCP) 443 | nexmo://ApiKey:ApiSecret@FromPhoneNo
nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo
nexmo://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ diff --git a/apprise/URLBase.py b/apprise/URLBase.py index 1cea66d1..3c969753 100644 --- a/apprise/URLBase.py +++ b/apprise/URLBase.py @@ -550,7 +550,7 @@ class URLBase: return paths @staticmethod - def parse_list(content, unquote=True): + def parse_list(content, allow_whitespace=True, unquote=True): """A wrapper to utils.parse_list() with unquoting support Parses a specified set of data and breaks it into a list. @@ -559,6 +559,9 @@ class URLBase: content (str): The path to split up into a list. If a list is provided, then it's individual entries are processed. + allow_whitespace (:obj:`bool`, optional): whitespace is to be + treated as a delimiter + unquote (:obj:`bool`, optional): call unquote on each element added to the returned list. @@ -566,7 +569,7 @@ class URLBase: list: A unique list containing all of the elements in the path """ - content = parse_list(content) + content = parse_list(content, allow_whitespace=allow_whitespace) if unquote: content = \ [URLBase.unquote(x) for x in filter(bool, content)] diff --git a/apprise/plugins/NotifyThreema.py b/apprise/plugins/NotifyThreema.py new file mode 100644 index 00000000..719401f6 --- /dev/null +++ b/apprise/plugins/NotifyThreema.py @@ -0,0 +1,370 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +# Create an account https://gateway.threema.ch/en/ if you don't already have +# one +# +# Read more about Threema Gateway API here: +# - https://gateway.threema.ch/en/developer/api + +import requests +from itertools import chain + +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..utils import is_phone_no +from ..utils import validate_regex +from ..utils import is_email +from ..URLBase import PrivacyMode +from ..utils import parse_list +from ..AppriseLocale import gettext_lazy as _ + + +class ThreemaRecipientTypes: + """ + The supported recipient specifiers + """ + THREEMA_ID = 'to' + PHONE = 'phone' + EMAIL = 'email' + + +class NotifyThreema(NotifyBase): + """ + A wrapper for Threema Gateway Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'Threema Gateway' + + # The services URL + service_url = 'https://gateway.threema.ch/' + + # The default protocol + secure_protocol = 'threema' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_threema' + + # Threema Gateway uses the http protocol with JSON requests + notify_url = 'https://msgapi.threema.ch/send_simple' + + # The maximum length of the body + body_maxlen = 3500 + + # No title support + title_maxlen = 0 + + # Define object templates + templates = ( + '{schema}://{gateway_id}@{secret}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'gateway_id': { + 'name': _('Gateway ID'), + 'type': 'string', + 'private': True, + 'required': True, + 'map_to': 'user', + }, + 'secret': { + 'name': _('API Secret'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'target_phone': { + 'name': _('Target Phone No'), + 'type': 'string', + 'prefix': '+', + 'regex': (r'^[0-9\s)(+-]+$', 'i'), + 'map_to': 'targets', + }, + 'target_email': { + 'name': _('Target Email'), + 'type': 'string', + 'map_to': 'targets', + }, + 'target_threema_id': { + 'name': _('Target Threema ID'), + 'type': 'string', + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + 'required': True, + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'to': { + 'alias_of': 'targets', + }, + 'from': { + 'alias_of': 'gateway_id', + }, + 'gwid': { + 'alias_of': 'gateway_id', + }, + 'secret': { + 'alias_of': 'secret', + }, + }) + + def __init__(self, secret=None, targets=None, **kwargs): + """ + Initialize Threema Gateway Object + """ + super().__init__(**kwargs) + + # Validate our params here. + + if not self.user: + msg = 'Threema Gateway ID must be specified' + self.logger.warning(msg) + raise TypeError(msg) + + # Verify our Gateway ID + if len(self.user) != 8: + msg = 'Threema Gateway ID must be 8 characters in length' + self.logger.warning(msg) + raise TypeError(msg) + + # Verify our secret + self.secret = validate_regex(secret) + if not self.secret: + msg = \ + 'An invalid Threema API Secret ({}) was specified'.format( + secret) + self.logger.warning(msg) + raise TypeError(msg) + + # Parse our targets + self.targets = list() + + # Used for URL generation afterwards only + self.invalid_targets = list() + + for target in parse_list(targets, allow_whitespace=False): + if len(target) == 8: + # Store our user + self.targets.append( + (ThreemaRecipientTypes.THREEMA_ID, target)) + continue + + # Check if an email was defined + result = is_email(target) + if result: + # Store our user + self.targets.append( + (ThreemaRecipientTypes.EMAIL, result['full_email'])) + continue + + # Validate targets and drop bad ones: + result = is_phone_no(target) + if result: + # store valid phone number + self.targets.append(( + ThreemaRecipientTypes.PHONE, result['full'])) + continue + + self.logger.warning( + 'Dropped invalid user/email/phone ' + '({}) specified'.format(target), + ) + self.invalid_targets.append(target) + + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Threema Gateway Notification + """ + + if len(self.targets) == 0: + # There were no services to notify + self.logger.warning( + 'There were no Threema Gateway targets to notify') + return False + + # error tracking (used for function return) + has_error = False + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8', + 'Accept': '*/*', + } + + # Prepare our payload + _payload = { + 'secret': self.secret, + 'from': self.user, + 'text': body.encode('utf-8'), + } + + # Create a copy of the targets list + targets = list(self.targets) + + while len(targets): + # Get our target to notify + key, target = targets.pop(0) + + # Prepare a payload object + payload = _payload.copy() + + # Set Target + payload[key] = target + + # Some Debug Logging + self.logger.debug( + 'Threema Gateway GET URL: {} (cert_verify={})'.format( + self.notify_url, self.verify_certificate)) + self.logger.debug('Threema Gateway Payload: {}' .format(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + self.notify_url, + params=payload, + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyThreema.http_response_code_lookup( + r.status_code) + + self.logger.warning( + 'Failed to send Threema Gateway notification to {}: ' + '{}{}error={}'.format( + target, + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + # Mark our failure + has_error = True + continue + + # We wee successful + self.logger.info( + 'Sent Threema Gateway notification to %s' % target) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending Threema Gateway:%s ' + 'notification' % target + ) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Mark our failure + has_error = True + continue + + return not has_error + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Define any URL parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + schemaStr = \ + '{schema}://{gatewayid}@{secret}/{targets}?{params}' + return schemaStr.format( + schema=self.secure_protocol, + gatewayid=NotifyThreema.quote(self.user), + secret=self.pprint( + self.secret, privacy, mode=PrivacyMode.Secret, safe=''), + targets='/'.join(chain( + [NotifyThreema.quote(x[1], safe='@+') for x in self.targets], + [NotifyThreema.quote(x, safe='@+') + for x in self.invalid_targets])), + params=NotifyThreema.urlencode(params)) + + def __len__(self): + """ + Returns the number of targets associated with this notification + """ + targets = len(self.targets) + return targets if targets > 0 else 1 + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + """ + + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't load the results + return results + + results['targets'] = list() + + if 'secret' in results['qsd'] and len(results['qsd']['secret']): + results['secret'] = \ + NotifyThreema.unquote(results['qsd']['secret']) + + else: + results['secret'] = NotifyThreema.unquote(results['host']) + + results['targets'] += \ + NotifyThreema.split_path(results['fullpath']) + + if 'from' in results['qsd'] and len(results['qsd']['from']): + results['user'] = \ + NotifyThreema.unquote(results['qsd']['from']) + + elif 'gwid' in results['qsd'] and len(results['qsd']['gwid']): + results['user'] = \ + NotifyThreema.unquote(results['qsd']['gwid']) + + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifyThreema.parse_list( + results['qsd']['to'], allow_whitespace=False) + + return results diff --git a/apprise/utils.py b/apprise/utils.py index 8d644ce9..af84f89d 100644 --- a/apprise/utils.py +++ b/apprise/utils.py @@ -96,6 +96,9 @@ VALID_QUERY_RE = re.compile(r'^(?P.*[/\\])(?P[^/\\]+)?$') # This is useful when turning a string into a list STRING_DELIMITERS = r'[\[\]\;,\s]+' +# String Delimiters without the whitespace +STRING_DELIMITERS_NO_WS = r'[\[\]\;,]+' + # Pre-Escape content since we reference it so much ESCAPED_PATH_SEPARATOR = re.escape('\\/') ESCAPED_WIN_PATH_SEPARATOR = re.escape('\\') @@ -1116,7 +1119,7 @@ def urlencode(query, doseq=False, safe='', encoding=None, errors=None): errors=errors) -def parse_list(*args, cast=None): +def parse_list(*args, cast=None, allow_whitespace=True): """ Take a string list and break it into a delimited list of arguments. This funciton also supports @@ -1143,10 +1146,12 @@ def parse_list(*args, cast=None): arg = cast(arg) if isinstance(arg, str): - result += re.split(STRING_DELIMITERS, arg) + result += re.split( + STRING_DELIMITERS if allow_whitespace + else STRING_DELIMITERS_NO_WS, arg) elif isinstance(arg, (set, list, tuple)): - result += parse_list(*arg) + result += parse_list(*arg, allow_whitespace=allow_whitespace) # # filter() eliminates any empty entries @@ -1154,7 +1159,9 @@ def parse_list(*args, cast=None): # Since Python v3 returns a filter (iterator) whereas Python v2 returned # a list, we need to change it into a list object to remain compatible with # both distribution types. - return sorted([x for x in filter(bool, list(set(result)))]) + return sorted([x for x in filter(bool, list(set(result)))]) \ + if allow_whitespace else sorted( + [x.strip() for x in filter(bool, list(set(result))) if x.strip()]) def is_exclusive_match(logic, data, match_all=common.MATCH_ALL_TAG, diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 00c81cdb..f51a1063 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -49,8 +49,8 @@ OneSignal, Opsgenie, PagerDuty, PagerTree, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, SparkPost, Super Toasty, -Streamlabs, Stride, Syslog, Techulus Push, Telegram, Twilio, Twitter, Twist, -XBMC, Voipms, Vonage, WhatsApp, Webex Teams} +Streamlabs, Stride, Syslog, Techulus Push, Telegram, Threema Gateway, Twilio, +Twitter, Twist, XBMC, Voipms, Vonage, WhatsApp, Webex Teams} Name: python-%{pypi_name} Version: 1.6.0 diff --git a/test/test_plugin_threema.py b/test/test_plugin_threema.py new file mode 100644 index 00000000..eb6da871 --- /dev/null +++ b/test/test_plugin_threema.py @@ -0,0 +1,178 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +from unittest import mock + +import pytest +import requests + +from apprise.plugins.NotifyThreema import NotifyThreema +from helpers import AppriseURLTester + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ('threema://', { + # No user/secret specified + 'instance': TypeError, + }), + ('threema://@:', { + # Invalid url + 'instance': TypeError, + }), + ('threema://user@secret', { + # gateway id must be 8 characters in len + 'instance': TypeError, + }), + ('threema://*THEGWID@secret/{targets}/'.format( + targets='/'.join(['2222'])), { + + # Invalid target phone number + 'instance': NotifyThreema, + 'notify_response': False, + 'privacy_url': 'threema://%2ATHEGWID@****/2222', + }), + ('threema://*THEGWID@secret/{targets}/'.format( + targets='/'.join(['16134442222'])), { + + # Valid + 'instance': NotifyThreema, + 'privacy_url': 'threema://%2ATHEGWID@****/16134442222', + }), + ('threema://*THEGWID@secret/{targets}/'.format( + targets='/'.join(['16134442222', '16134443333'])), { + + # Valid multiple targets + 'instance': NotifyThreema, + 'privacy_url': 'threema://%2ATHEGWID@****/16134442222/16134443333', + }), + ('threema:///?secret=secret&from=*THEGWID&to={targets}'.format( + targets=','.join(['16134448888', 'user1@gmail.com', 'abcd1234'])), { + + # Valid + 'instance': NotifyThreema, + }), + ('threema:///?secret=secret&gwid=*THEGWID&to={targets}'.format( + targets=','.join(['16134448888', 'user2@gmail.com', 'abcd1234'])), { + + # Valid + 'instance': NotifyThreema, + }), + ('threema://*THEGWID@secret', { + 'instance': NotifyThreema, + # No targets specified + 'notify_response': False, + }), + + ('threema://*THEGWID@secret/16134443333', { + 'instance': NotifyThreema, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('threema://*THEGWID@secret/16134443333', { + 'instance': NotifyThreema, + # Throws a series of errors + 'test_requests_exceptions': True, + }), +) + + +def test_plugin_threema(): + """ + NotifyThreema() Apprise URLs + + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@ mock.patch('requests.post') +def test_plugin_threema_edge_cases(mock_post): + """ + NotifyThreema() Edge Cases + + """ + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + + # Initialize some generic (but valid) tokens + gwid = '*THEGWID' + secret = 'mysecret' + targets = '+1 (555) 123-9876' + + # No email specified + with pytest.raises(TypeError): + NotifyThreema(user=gwid, secret=None, targets=targets) + + results = NotifyThreema.parse_url( + f'threema://?gwid={gwid}&secret={secret}&to={targets}') + + assert isinstance(results, dict) + assert results['user'] == gwid + assert results['secret'] == secret + assert results['password'] is None + assert results['port'] is None + assert results['host'] == '' + assert results['fullpath'] == '/' + assert results['path'] == '/' + assert results['query'] is None + assert results['schema'] == 'threema' + assert results['url'] == 'threema:///' + assert isinstance(results['targets'], list) is True + assert len(results['targets']) == 1 + assert results['targets'][0] == '+1 (555) 123-9876' + + instance = NotifyThreema(**results) + assert len(instance.targets) == 1 + assert instance.targets[0] == ('phone', '15551239876') + assert isinstance(instance, NotifyThreema) + + response = instance.send(title='title', body='body 😊') + assert response is True + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://msgapi.threema.ch/send_simple' + assert details[1]['headers']['User-Agent'] == 'Apprise' + assert details[1]['headers']['Accept'] == '*/*' + assert details[1]['headers']['Content-Type'] == \ + 'application/x-www-form-urlencoded; charset=utf-8' + assert details[1]['params']['secret'] == secret + assert details[1]['params']['from'] == gwid + assert details[1]['params']['phone'] == '15551239876' + assert details[1]['params']['text'] == 'body 😊'.encode('utf-8')