Added SMS Manager Support (#1018)

pull/1027/head
Chris Caron 11 months ago committed by GitHub
parent cda692594d
commit 2228bda1dd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -81,6 +81,7 @@ SimplePush
Sinch
Slack
SMSEagle
SMS Manager
SMTP2Go
SNS
SparkPost

@ -150,6 +150,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<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo<br/>signal://hostname:port/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Sinch](https://github.com/caronc/apprise/wiki/Notify_sinch) | sinch:// | (TCP) 443 | sinch://ServicePlanId:ApiToken@FromPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo<br/>sinch://ServicePlanId:ApiToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>sinch://ServicePlanId:ApiToken@ShortCode/ToPhoneNo<br/>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<br/>smseagles://hostname:port/@ToContact<br/>smseagles://hostname:port/#ToGroup<br/>smseagles://hostname:port/ToPhoneNo1/#ToGroup/@ToContact/
[SMS Manager](https://github.com/caronc/apprise/wiki/Notify_sms_manager) | smsmgr:// | (TCP) 443 | smsmgr://ApiKey@ToPhoneNo<br/>smsmgr://ApiKey@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Threema Gateway](https://github.com/caronc/apprise/wiki/Notify_threema) | threema:// | (TCP) 443 | threema://GatewayID@secret/ToPhoneNo<br/>threema://GatewayID@secret/ToEmail<br/>threema://GatewayID@secret/ToThreemaID/<br/>threema://GatewayID@secret/ToEmail/ToThreemaID/ToPhoneNo/...
| [Twilio](https://github.com/caronc/apprise/wiki/Notify_twilio) | twilio:// | (TCP) 443 | twilio://AccountSid:AuthToken@FromPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/<br/>twilio://AccountSid:AuthToken@FromPhoneNo/ToPhoneNo?apikey=Key<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo<br/>twilio://AccountSid:AuthToken@ShortCode/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Voipms](https://github.com/caronc/apprise/wiki/Notify_voipms) | voipms:// | (TCP) 443 | voipms://password:email/FromPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo<br/>voipms://password:email/FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/

@ -0,0 +1,413 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# 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.
# API Reference: https://smsmanager.cz/api/http#send
# To use this service you will need a SMS Manager account
# You will need credits (new accounts start with a few)
# https://smsmanager.cz
# 1. Sign up and get test credit
# 2. Generate an API key in web administration.
import requests
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import is_phone_no
from ..utils import parse_phone_no
from ..utils import parse_bool
from ..utils import validate_regex
from ..AppriseLocale import gettext_lazy as _
class SMSManagerGateway(object):
"""
The different gateway values
"""
HIGH = "high"
ECONOMY = "economy"
LOW = "low"
DIRECT = "direct"
# Used for verification purposes
SMS_MANAGER_GATEWAYS = (
SMSManagerGateway.HIGH,
SMSManagerGateway.ECONOMY,
SMSManagerGateway.LOW,
SMSManagerGateway.DIRECT,
)
class NotifySMSManager(NotifyBase):
"""
A wrapper for SMS Manager Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'SMS Manager'
# The services URL
service_url = 'https://smsmanager.cz'
# All notification requests are secure
secure_protocol = ('smsmgr', 'smsmanager',)
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_sms_manager'
# SMS Manager uses the http protocol with JSON requests
notify_url = 'https://http-api.smsmanager.cz/Send'
# The maximum amount of texts that can go out in one batch
default_batch_size = 4000
# The maximum length of the body
body_maxlen = 160
# A title can not be used for SMS Messages. Setting this to zero will
# cause any title (if defined) to get placed into the message body.
title_maxlen = 0
# Define object templates
templates = (
'{schema}://{apikey}@{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'private': True,
'required': True,
},
'target_phone': {
'name': _('Target Phone No'),
'type': 'string',
'prefix': '+',
'regex': (r'^[0-9\s)(+-]+$', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
'required': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'key': {
'alias_of': 'apikey',
},
'to': {
'alias_of': 'targets',
},
'from': {
'name': _('From Phone No'),
'type': 'string',
'regex': (r'^\+?[0-9\s)(+-]+$', 'i'),
'map_to': 'sender',
},
'sender': {
'alias_of': 'from',
},
'gateway': {
'name': _('Gateway'),
'type': 'choice:string',
'values': SMS_MANAGER_GATEWAYS,
'default': SMS_MANAGER_GATEWAYS[0],
},
'batch': {
'name': _('Batch Mode'),
'type': 'bool',
'default': False,
},
})
def __init__(self, apikey=None, sender=None, targets=None, batch=None,
gateway=None, **kwargs):
"""
Initialize SMS Manager Object
"""
super(NotifySMSManager, self).__init__(**kwargs)
self.apikey = validate_regex(apikey)
if not self.apikey:
msg = 'An invalid API Key ({}) was specified.'.format(apikey)
self.logger.warning(msg)
raise TypeError(msg)
# Setup our gateway
self.gateway = self.template_args['gateway']['default'] \
if not isinstance(gateway, str) else gateway.lower()
if self.gateway not in SMS_MANAGER_GATEWAYS:
msg = 'The Gateway specified ({}) is invalid.'.format(gateway)
self.logger.warning(msg)
raise TypeError(msg)
# Define whether or not we should operate in a batch mode
self.batch = self.template_args['batch']['default'] \
if batch is None else bool(batch)
# Maximum 11 characters and must be approved by administrators of site
self.sender = sender[0:11] if isinstance(sender, str) else None
# Parse our targets
self.targets = list()
for target in parse_phone_no(targets):
# Parse each phone number we found
# It is documented that numbers with a length of 9 characters are
# supplemented by "420".
result = is_phone_no(target, min_len=9)
if result:
# Carry forward '+' if defined, otherwise do not...
self.targets.append(
('+' + result['full'])
if target.lstrip()[0] == '+' else result['full'])
continue
self.logger.warning(
'Dropped invalid phone # ({}) specified.'.format(target),
)
return
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform SMS Manager Notification
"""
if not self.targets:
# We have nothing to notify
self.logger.warning('There are no SMS Manager targets to notify')
return False
# error tracking (used for function return)
has_error = False
# Send in batches if identified to do so
batch_size = 1 if not self.batch else self.default_batch_size
# Prepare our headers
headers = {
'User-Agent': self.app_id,
}
# Prepare our targets
targets = list(self.targets) if batch_size == 1 else \
[self.targets[index:index + batch_size]
for index in range(0, len(self.targets), batch_size)]
while len(targets):
# Get our target to notify
target = targets.pop(0)
# Prepare our payload
# Note: Payload is assembled inside of our while-loop due to
# mock testing issues (payload singleton isn't persistent
# when performing follow up checks on the params object.
payload = {
'apikey': self.apikey,
'gateway': self.gateway,
# The number gets populated in the loop below
'number': None,
'message': body,
}
if self.sender:
# Sender is ony set if specified
payload['sender'] = self.sender
# Printable target details
if isinstance(target, list):
p_target = '{} targets'.format(len(target))
# Prepare our target numbers
payload['number'] = ';'.join(target)
else:
p_target = target
# Prepare our target numbers
payload['number'] = target
# Some Debug Logging
self.logger.debug(
'SMS Manager POST URL: {} (cert_verify={})'.format(
self.notify_url, self.verify_certificate))
self.logger.debug('SMS Manager Payload: {}' .format(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.get(
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 = \
NotifyBase.http_response_code_lookup(r.status_code)
# set up our status code to use
status_code = r.status_code
self.logger.warning(
'Failed to send SMS Manager notification to {}: '
'{}{}error={}.'.format(
p_target,
status_str,
', ' if status_str else '',
status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
# Mark our failure
has_error = True
continue
else:
self.logger.info(
'Sent SMS Manager notification to {}.'.format(
p_target))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending SMS Manager: to %s ',
p_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 = {
'batch': 'yes' if self.batch else 'no',
'gateway': self.gateway,
}
if self.sender:
# Set our sender if it was set
params['sender'] = self.sender
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{apikey}@{targets}' \
'?{params}'.format(
schema=self.secure_protocol[0],
apikey=self.pprint(self.apikey, privacy, safe=''),
targets='/'.join([
NotifySMSManager.quote('{}'.format(x), safe='+')
for x in self.targets]),
params=NotifySMSManager.urlencode(params))
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
#
# Factor batch into calculation
#
# Note: Groups always require a separate request (and can not be
# included in batch calculations)
batch_size = 1 if not self.batch else self.default_batch_size
targets = len(self.targets)
if batch_size > 1:
targets = int(targets / batch_size) + \
(1 if targets % batch_size else 0)
return targets
@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
# Get our API Key
results['apikey'] = NotifySMSManager.unquote(results['user'])
# Store our targets
results['targets'] = [
*NotifySMSManager.parse_phone_no(results['host']),
*NotifySMSManager.split_path(results['fullpath'])]
# The 'from' makes it easier to use yaml configuration
if 'from' in results['qsd'] and len(results['qsd']['from']):
results['sender'] = \
NotifySMSManager.unquote(results['qsd']['from'])
elif 'sender' in results['qsd'] and len(results['qsd']['sender']):
# Support sender= value as well to align with SMS Manager API
results['sender'] = \
NotifySMSManager.unquote(results['qsd']['sender'])
# Support the 'to' variable so that we can support targets this way too
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifySMSManager.parse_phone_no(results['qsd']['to'])
if 'key' in results['qsd'] and len(results['qsd']['key']):
results['apikey'] = \
NotifySMSManager.unquote(results['qsd']['key'])
# Get Batch Mode Flag
results['batch'] = \
parse_bool(results['qsd'].get(
'batch', NotifySMSManager.template_args['batch']['default']))
# Define our gateway
if 'gateway' in results['qsd'] and len(results['qsd']['gateway']):
results['gateway'] = \
NotifySMSManager.unquote(results['qsd']['gateway'])
return results

@ -48,8 +48,8 @@ MQTT, MSG91, MyAndroid, Nexmo, Nextcloud, NextcloudTalk, Notica, Notifiarr,
Notifico, ntfy, Office365, 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, Synology Chat, Syslog,
ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go,
Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog,
Techulus Push, Telegram, Threema Gateway, Twilio, Twitter, Twist, XBMC,
Voipms, Vonage, WeCom Bot, WhatsApp, Webex Teams}

@ -0,0 +1,193 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# 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 requests
from apprise import Apprise
from apprise.plugins.NotifySMSManager import NotifySMSManager
from helpers import AppriseURLTester
from apprise import NotifyType
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Our Testing URLs
apprise_url_tests = (
('smsmgr://', {
# Instantiated but no auth, so no otification can happen
'instance': TypeError,
}),
('smsmgr://:@/', {
# invalid auth
'instance': TypeError
}),
('smsmgr://{}@{}'.format('b' * 10, '3' * 5), {
# invalid nubmer provided
'instance': NotifySMSManager,
# Expected notify() response because we have no one to notify
'notify_response': False,
}),
('smsmgr://{}@123/{}/abcd/+{}'.format(
'z' * 10, '3' * 11, '4' * 11), {
# includes a few invalid bits of info
'instance': NotifySMSManager,
'privacy_url': 'smsmgr://z...z@33333333333/+44444444444'
}),
('smsmgr://{}@{}?batch=y'.format(
'b' * 5, '4' * 11), {
'instance': NotifySMSManager,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'smsmgr://b...b@44444444444',
}),
# Test gateway group
('smsmgr://{}@{}?gateway=low'.format('a' * 10, '1' * 11), {
'instance': NotifySMSManager,
}),
('smsmgr://{}@{}?gateway=invalid'.format('a' * 10, '1' * 11), {
# invalid gatewwway
'instance': TypeError,
}),
('smsmgr://{}?key={}&from=user'.format('1' * 11, 'a' * 10), {
# use get args to acomplish the same thing
'instance': NotifySMSManager,
}),
('smsmgr://_?to={},{}&key={}&sender={}'.format(
'1' * 11, '2' * 11, 'b' * 10, '5' * 13), {
# use get args to acomplish the same thing
'instance': NotifySMSManager,
}),
('smsmgr://{}@{}'.format('a' * 10, '1' * 11), {
'instance': NotifySMSManager,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('smsmgr://{}@{}'.format('a' * 10, '1' * 11), {
'instance': NotifySMSManager,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
)
def test_plugin_smsmgr_urls():
"""
NotifyTemplate() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@mock.patch('requests.get')
def test_plugin_smsmgr_edge_cases(mock_get):
"""
NotifySMSManager() Edge Cases
"""
# Initialize some generic (but valid) tokens
apikey = 'my-api-key'
targets = [
'+1(555) 123-1234',
'1555 5555555',
# A garbage entry
'12',
# NOw a valid one because a group was implicit
'@12',
]
# Prepare our response
response = requests.Request()
response.status_code = requests.codes.ok
# Prepare Mock
mock_get.return_value = response
# Instantiate our object
obj = Apprise.instantiate(
'smsmgr://{}@{}?batch=n'.format(apikey, '/'.join(targets)))
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# We know there are 2 (valid) targets
assert len(obj) == 2
# Test our call count
assert mock_get.call_count == 2
# Test
details = mock_get.call_args_list[0]
payload = details[1]['params']
assert payload['apikey'] == apikey
assert payload['gateway'] == 'high'
assert payload['number'] == '+15551231234'
assert payload['message'] == 'title\r\nbody'
details = mock_get.call_args_list[1]
payload = details[1]['params']
assert payload['apikey'] == apikey
assert payload['gateway'] == 'high'
assert payload['number'] == '15555555555'
assert payload['message'] == 'title\r\nbody'
# Verify our URL looks good
assert obj.url().startswith(
'smsmgr://{}@{}'.format(apikey, '/'.join(
['+15551231234', '15555555555'])))
assert 'batch=no' in obj.url()
# Reset our mock object
mock_get.reset_mock()
# With our batch in place, our calculations are different
obj = Apprise.instantiate(
'smsmgr://{}@{}?batch=y'.format(apikey, '/'.join(targets)))
# 2 phones were loaded but counted as 1 due to batch flag
assert len(obj) == 1
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Test our call count (batched into 1)
assert mock_get.call_count == 1
details = mock_get.call_args_list[0]
payload = details[1]['params']
assert payload['apikey'] == apikey
assert payload['gateway'] == 'high'
assert payload['number'] == '+15551231234;15555555555'
assert payload['message'] == 'title\r\nbody'
Loading…
Cancel
Save