Burst SMS Support Added (#898)

This commit is contained in:
Chris Caron 2023-07-01 16:38:52 -04:00 committed by GitHub
parent f1d0e9b5ec
commit e0f42d291a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 656 additions and 2 deletions

View File

@ -3,6 +3,7 @@ API
AWS
Boxcar
BulkSMS
Burst SMS
Chat
CLI
ClickSend

View File

@ -130,6 +130,7 @@ The table below identifies the services this tool supports and some example serv
| -------------------- | ---------- | ------------ | -------------- |
| [AWS SNS](https://github.com/caronc/apprise/wiki/Notify_sns) | sns:// | (TCP) 443 | sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo<br/>sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo1/+PhoneNo2/+PhoneNoN<br/>sns://AccessKeyID/AccessSecretKey/RegionName/Topic<br/>sns://AccessKeyID/AccessSecretKey/RegionName/Topic1/Topic2/TopicN
| [BulkSMS](https://github.com/caronc/apprise/wiki/Notify_bulksms) | bulksms:// | (TCP) 443 | bulksms://user:password@ToPhoneNo<br/>bulksms://User:Password@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [Burst SMS](https://github.com/caronc/apprise/wiki/Notify_burst_sms) | burstsms:// | (TCP) 443 | burstsms://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo<br/>burstsms://ApiKey:ApiSecret@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
| [ClickSend](https://github.com/caronc/apprise/wiki/Notify_clicksend) | clicksend:// | (TCP) 443 | clicksend://user:pass@PhoneNo<br/>clicksend://user:pass@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN
| [DAPNET](https://github.com/caronc/apprise/wiki/Notify_dapnet) | dapnet:// | (TCP) 80 | dapnet://user:pass@callsign<br/>dapnet://user:pass@callsign1/callsign2/callsignN
| [D7 Networks](https://github.com/caronc/apprise/wiki/Notify_d7networks) | d7sms:// | (TCP) 443 | d7sms://token@PhoneNo<br/>d7sms://token@ToPhoneNo1/ToPhoneNo2/ToPhoneNoN

View File

@ -0,0 +1,464 @@
# -*- coding: utf-8 -*-
# BSD 3-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.
#
# 3. Neither the name of the copyright holder nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# 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.
# Sign-up with https://burstsms.com/
#
# Define your API Secret here and acquire your API Key
# - https://can.transmitsms.com/profile
#
import requests
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
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 BurstSMSCountryCode:
# Australia
AU = 'au'
# New Zeland
NZ = 'nz'
# United Kingdom
UK = 'gb'
# United States
US = 'us'
BURST_SMS_COUNTRY_CODES = (
BurstSMSCountryCode.AU,
BurstSMSCountryCode.NZ,
BurstSMSCountryCode.UK,
BurstSMSCountryCode.US,
)
class NotifyBurstSMS(NotifyBase):
"""
A wrapper for Burst SMS Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Burst SMS'
# The services URL
service_url = 'https://burstsms.com/'
# The default protocol
secure_protocol = 'burstsms'
# The maximum amount of SMS Messages that can reside within a single
# batch transfer based on:
# https://developer.transmitsms.com/#74911cf8-dec6-4319-a499-7f535a7fd08c
default_batch_size = 500
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_burst_sms'
# Burst SMS uses the http protocol with JSON requests
notify_url = 'https://api.transmitsms.com/send-sms.json'
# 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}:{secret}@{sender_id}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'required': True,
'regex': (r'^[a-z0-9]+$', 'i'),
'private': True,
},
'secret': {
'name': _('API Secret'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'^[a-z0-9]+$', 'i'),
},
'sender_id': {
'name': _('Sender ID'),
'type': 'string',
'required': True,
'map_to': 'source',
},
'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, **{
'to': {
'alias_of': 'targets',
},
'from': {
'alias_of': 'sender_id',
},
'key': {
'alias_of': 'apikey',
},
'secret': {
'alias_of': 'secret',
},
'country': {
'name': _('Country'),
'type': 'choice:string',
'values': BURST_SMS_COUNTRY_CODES,
'default': BurstSMSCountryCode.US,
},
# Validity
# Expire a message send if it is undeliverable (defined in minutes)
# If set to Zero (0); this is the default and sets the max validity
# period
'validity': {
'name': _('validity'),
'type': 'int',
'default': 0
},
'batch': {
'name': _('Batch Mode'),
'type': 'bool',
'default': False,
},
})
def __init__(self, apikey, secret, source, targets=None, country=None,
validity=None, batch=None, **kwargs):
"""
Initialize Burst SMS Object
"""
super().__init__(**kwargs)
# API Key (associated with project)
self.apikey = validate_regex(
apikey, *self.template_tokens['apikey']['regex'])
if not self.apikey:
msg = 'An invalid Burst SMS API Key ' \
'({}) was specified.'.format(apikey)
self.logger.warning(msg)
raise TypeError(msg)
# API Secret (associated with project)
self.secret = validate_regex(
secret, *self.template_tokens['secret']['regex'])
if not self.secret:
msg = 'An invalid Burst SMS API Secret ' \
'({}) was specified.'.format(secret)
self.logger.warning(msg)
raise TypeError(msg)
if not country:
self.country = self.template_args['country']['default']
else:
self.country = country.lower().strip()
if country not in BURST_SMS_COUNTRY_CODES:
msg = 'An invalid Burst SMS country ' \
'({}) was specified.'.format(country)
self.logger.warning(msg)
raise TypeError(msg)
# Set our Validity
self.validity = self.template_args['validity']['default']
if validity:
try:
self.validity = int(validity)
except (ValueError, TypeError):
msg = 'The Burst SMS Validity specified ({}) is invalid.'\
.format(validity)
self.logger.warning(msg)
raise TypeError(msg)
# Prepare Batch Mode Flag
self.batch = self.template_args['batch']['default'] \
if batch is None else batch
# The Sender ID
self.source = validate_regex(source)
if not self.source:
msg = 'The Account Sender ID specified ' \
'({}) is invalid.'.format(source)
self.logger.warning(msg)
raise TypeError(msg)
# Parse our targets
self.targets = list()
for target in parse_phone_no(targets):
# Validate targets and drop bad ones:
result = is_phone_no(target)
if not result:
self.logger.warning(
'Dropped invalid phone # '
'({}) specified.'.format(target),
)
continue
# store valid phone number
self.targets.append(result['full'])
return
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Burst SMS Notification
"""
if not self.targets:
self.logger.warning(
'There are no valid Burst SMS targets to notify.')
return False
# error tracking (used for function return)
has_error = False
# Prepare our headers
headers = {
'User-Agent': self.app_id,
'Accept': 'application/json',
}
# Prepare our authentication
auth = (self.apikey, self.secret)
# Prepare our payload
payload = {
'countrycode': self.country,
'message': body,
# Sender ID
'from': self.source,
# The to gets populated in the loop below
'to': None,
}
# Send in batches if identified to do so
batch_size = 1 if not self.batch else self.default_batch_size
# Create a copy of the targets list
targets = list(self.targets)
for index in range(0, len(targets), batch_size):
# Prepare our user
payload['to'] = ','.join(self.targets[index:index + batch_size])
# Some Debug Logging
self.logger.debug('Burst SMS POST URL: {} (cert_verify={})'.format(
self.notify_url, self.verify_certificate))
self.logger.debug('Burst SMS Payload: {}' .format(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.notify_url,
data=payload,
headers=headers,
auth=auth,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyBurstSMS.http_response_code_lookup(
r.status_code)
self.logger.warning(
'Failed to send Burst SMS notification to {} '
'target(s): {}{}error={}.'.format(
len(self.targets[index:index + batch_size]),
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
else:
self.logger.info(
'Sent Burst SMS notification to %d target(s).' %
len(self.targets[index:index + batch_size]))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending Burst SMS '
'notification to %d target(s).' %
len(self.targets[index:index + batch_size]))
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 = {
'country': self.country,
'batch': 'yes' if self.batch else 'no',
}
if self.validity:
params['validity'] = str(self.validity)
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
return '{schema}://{key}:{secret}@{source}/{targets}/?{params}'.format(
schema=self.secure_protocol,
key=self.pprint(self.apikey, privacy, safe=''),
secret=self.pprint(
self.secret, privacy, mode=PrivacyMode.Secret, safe=''),
source=NotifyBurstSMS.quote(self.source, safe=''),
targets='/'.join(
[NotifyBurstSMS.quote(x, safe='') for x in self.targets]),
params=NotifyBurstSMS.urlencode(params))
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
#
# Factor batch into calculation
#
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 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
# The hostname is our source (Sender ID)
results['source'] = NotifyBurstSMS.unquote(results['host'])
# Get any remaining targets
results['targets'] = NotifyBurstSMS.split_path(results['fullpath'])
# Get our account_side and auth_token from the user/pass config
results['apikey'] = NotifyBurstSMS.unquote(results['user'])
results['secret'] = NotifyBurstSMS.unquote(results['password'])
# API Key
if 'key' in results['qsd'] and len(results['qsd']['key']):
# Extract the API Key from an argument
results['apikey'] = \
NotifyBurstSMS.unquote(results['qsd']['key'])
# API Secret
if 'secret' in results['qsd'] and len(results['qsd']['secret']):
# Extract the API Secret from an argument
results['secret'] = \
NotifyBurstSMS.unquote(results['qsd']['secret'])
# Support the 'from' and 'source' variable so that we can support
# targets this way too.
# The 'from' makes it easier to use yaml configuration
if 'from' in results['qsd'] and len(results['qsd']['from']):
results['source'] = \
NotifyBurstSMS.unquote(results['qsd']['from'])
if 'source' in results['qsd'] and len(results['qsd']['source']):
results['source'] = \
NotifyBurstSMS.unquote(results['qsd']['source'])
# Support country
if 'country' in results['qsd'] and len(results['qsd']['country']):
results['country'] = \
NotifyBurstSMS.unquote(results['qsd']['country'])
# Support validity value
if 'validity' in results['qsd'] and len(results['qsd']['validity']):
results['validity'] = \
NotifyBurstSMS.unquote(results['qsd']['validity'])
# Get Batch Mode Flag
if 'batch' in results['qsd'] and len(results['qsd']['batch']):
results['batch'] = parse_bool(results['qsd']['batch'])
# Support the 'to' variable so that we can support rooms this way too
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyBurstSMS.parse_phone_no(results['qsd']['to'])
return results

View File

@ -43,8 +43,8 @@ Apprise is a Python package for simplifying access to all of the different
notification services that are out there. Apprise opens the door and makes
it easy to access:
Apprise API, AWS SES, AWS SNS, Bark, BulkSMS, Boxcar, ClickSend, DAPNET,
DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter, Google Chat,
Apprise API, AWS SES, AWS SNS, Bark, Boxcar, Burst SMS, BulkSMS, ClickSend,
DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter, Google Chat,
Gotify, Growl, Guilded, Home Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos,
LaMetric, Line, MacOSX, Mailgun, Mastodon, Mattermost, Matrix, MessageBird,
Microsoft Windows, Microsoft Teams, Misskey, MQTT, MSG91, MyAndroid, Nexmo,

View File

@ -0,0 +1,188 @@
# -*- coding: utf-8 -*-
# BSD 3-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.
#
# 3. Neither the name of the copyright holder nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# 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 json import dumps
from apprise.plugins.NotifyBurstSMS import NotifyBurstSMS
from helpers import AppriseURLTester
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Our Testing URLs
apprise_url_tests = (
('burstsms://', {
# No API Key specified
'instance': TypeError,
}),
('burstsms://:@/', {
# invalid Auth key
'instance': TypeError,
}),
('burstsms://{}@12345678'.format('a' * 8), {
# Just a key provided
'instance': TypeError,
}),
('burstsms://{}:{}@%20'.format('d' * 8, 'e' * 16), {
# Invalid source number
'instance': TypeError,
}),
('burstsms://{}:{}@{}/123/{}/abcd/'.format(
'f' * 8, 'g' * 16, '3' * 11, '9' * 15), {
# valid everything but target numbers
'instance': NotifyBurstSMS,
# Expected notify() response because not all targets are valid
'notify_response': False,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'burstsms://f...f:****@',
}),
('burstsms://{}:{}@{}'.format('h' * 8, 'i' * 16, '5' * 11), {
'instance': NotifyBurstSMS,
# Expected notify() response because no targets are defined
'notify_response': False,
}),
('burstsms://_?key={}&secret={}&from={}&to={}'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
# use get args to acomplish the same thing
'instance': NotifyBurstSMS,
}),
('burstsms://_?key={}&secret={}&from={}&to={}&batch=y'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
# batch flag set
'instance': NotifyBurstSMS,
}),
# Test our country
('burstsms://_?key={}&secret={}&source={}&to={}&country=us'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
'instance': NotifyBurstSMS,
}),
# Test an invalid country
('burstsms://_?key={}&secret={}&source={}&to={}&country=invalid'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
'instance': TypeError,
}),
# Test our validity
('burstsms://_?key={}&secret={}&source={}&to={}&validity=10'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
'instance': NotifyBurstSMS,
}),
# Test an invalid country
('burstsms://_?key={}&secret={}&source={}&to={}&validity=invalid'.format(
'a' * 8, 'b' * 16, '5' * 11, '6' * 11), {
'instance': TypeError,
}),
('burstsms://_?key={}&secret={}&from={}&to={}'.format(
'a' * 8, 'b' * 16, '5' * 11, '7' * 11), {
# use to=
'instance': NotifyBurstSMS,
}),
('burstsms://{}:{}@{}/{}'.format('a' * 8, 'b' * 16, '6' * 11, '7' * 11), {
'instance': NotifyBurstSMS,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('burstsms://{}:{}@{}/{}'.format('a' * 8, 'b' * 16, '6' * 11, '7' * 11), {
'instance': NotifyBurstSMS,
# 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_burstsms_urls():
"""
NotifyBurstSMS() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@mock.patch('requests.post')
def test_plugin_burstsms_edge_cases(mock_post):
"""
NotifyBurstSMS() 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
apikey = '{}'.format('b' * 8)
secret = '{}'.format('b' * 16)
source = '+1 (555) 123-3456'
# No apikey specified
with pytest.raises(TypeError):
NotifyBurstSMS(apikey=None, secret=secret, source=source)
with pytest.raises(TypeError):
NotifyBurstSMS(apikey=" ", secret=secret, source=source)
# No secret specified
with pytest.raises(TypeError):
NotifyBurstSMS(apikey=apikey, secret=None, source=source)
with pytest.raises(TypeError):
NotifyBurstSMS(apikey=apikey, secret=" ", source=source)
# a error response
response.status_code = 400
response.content = dumps({
"error": {
"code": "FIELD_INVALID",
"description":
"Sender ID must be one of the numbers that are currently leased.",
},
})
mock_post.return_value = response
# Initialize our object
obj = NotifyBurstSMS(apikey=apikey, secret=secret, source=source)
# We will fail with the above error code
assert obj.notify('title', 'body', 'info') is False