mirror of https://github.com/caronc/apprise
Chris Caron
3 years ago
committed by
GitHub
6 changed files with 646 additions and 10 deletions
@ -0,0 +1,400 @@
|
||||
# -*- coding: utf-8 -*- |
||||
# |
||||
# Copyright (C) 2020 Chris Caron <lead2gold@gmail.com> |
||||
# All rights reserved. |
||||
# |
||||
# This code is licensed under the MIT License. |
||||
# |
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy |
||||
# of this software and associated documentation files(the "Software"), to deal |
||||
# in the Software without restriction, including without limitation the rights |
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell |
||||
# copies of the Software, and to permit persons to whom the Software is |
||||
# furnished to do so, subject to the following conditions : |
||||
# |
||||
# The above copyright notice and this permission notice shall be included in |
||||
# all copies or substantial portions of the Software. |
||||
# |
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR |
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, |
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE |
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
||||
# 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 requests |
||||
from json import dumps |
||||
|
||||
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 ..URLBase import PrivacyMode |
||||
from ..AppriseLocale import gettext_lazy as _ |
||||
|
||||
|
||||
class NotifySignalAPI(NotifyBase): |
||||
""" |
||||
A wrapper for SignalAPI Notifications |
||||
""" |
||||
|
||||
# The default descriptive name associated with the Notification |
||||
service_name = 'Signal API' |
||||
|
||||
# The services URL |
||||
service_url = 'https://bbernhard.github.io/signal-cli-rest-api/' |
||||
|
||||
# The default protocol |
||||
protocol = 'signal' |
||||
|
||||
# The default protocol |
||||
secure_protocol = 'signals' |
||||
|
||||
# A URL that takes you to the setup/help of the specific protocol |
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_signal' |
||||
|
||||
# The maximum targets to include when doing batch transfers |
||||
default_batch_size = 10 |
||||
|
||||
# We don't support titles for Signal notifications |
||||
title_maxlen = 0 |
||||
|
||||
# Define object templates |
||||
templates = ( |
||||
'{schema}://{host}/{from_phone}', |
||||
'{schema}://{host}:{port}/{from_phone}', |
||||
'{schema}://{user}@{host}/{from_phone}', |
||||
'{schema}://{user}@{host}:{port}/{from_phone}', |
||||
'{schema}://{user}:{password}@{host}/{from_phone}', |
||||
'{schema}://{user}:{password}@{host}:{port}/{from_phone}', |
||||
'{schema}://{host}/{from_phone}/{targets}', |
||||
'{schema}://{host}:{port}/{from_phone}/{targets}', |
||||
'{schema}://{user}@{host}/{from_phone}/{targets}', |
||||
'{schema}://{user}@{host}:{port}/{from_phone}/{targets}', |
||||
'{schema}://{user}:{password}@{host}/{from_phone}/{targets}', |
||||
'{schema}://{user}:{password}@{host}:{port}/{from_phone}/{targets}', |
||||
) |
||||
|
||||
# Define our template tokens |
||||
template_tokens = dict(NotifyBase.template_tokens, **{ |
||||
'host': { |
||||
'name': _('Hostname'), |
||||
'type': 'string', |
||||
'required': True, |
||||
}, |
||||
'port': { |
||||
'name': _('Port'), |
||||
'type': 'int', |
||||
'min': 1, |
||||
'max': 65535, |
||||
}, |
||||
'user': { |
||||
'name': _('Username'), |
||||
'type': 'string', |
||||
}, |
||||
'password': { |
||||
'name': _('Password'), |
||||
'type': 'string', |
||||
'private': True, |
||||
}, |
||||
'from_phone': { |
||||
'name': _('From Phone No'), |
||||
'type': 'string', |
||||
'required': True, |
||||
'regex': (r'^\+?[0-9\s)(+-]+$', 'i'), |
||||
'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', |
||||
} |
||||
}) |
||||
|
||||
# Define our template arguments |
||||
template_args = dict(NotifyBase.template_args, **{ |
||||
'to': { |
||||
'alias_of': 'targets', |
||||
}, |
||||
'from': { |
||||
'alias_of': 'from_phone', |
||||
}, |
||||
'batch': { |
||||
'name': _('Batch Mode'), |
||||
'type': 'bool', |
||||
'default': False, |
||||
}, |
||||
'status': { |
||||
'name': _('Show Status'), |
||||
'type': 'bool', |
||||
'default': False, |
||||
}, |
||||
}) |
||||
|
||||
def __init__(self, source=None, targets=None, batch=False, status=False, |
||||
**kwargs): |
||||
""" |
||||
Initialize SignalAPI Object |
||||
""" |
||||
super(NotifySignalAPI, self).__init__(**kwargs) |
||||
|
||||
# Prepare Batch Mode Flag |
||||
self.batch = batch |
||||
|
||||
# Set Status type |
||||
self.status = status |
||||
|
||||
# Parse our targets |
||||
self.targets = list() |
||||
|
||||
# Used for URL generation afterwards only |
||||
self.invalid_targets = list() |
||||
|
||||
# Manage our Source Phone |
||||
result = is_phone_no(source) |
||||
if not result: |
||||
msg = 'An invalid Signal API Source Phone No ' \ |
||||
'({}) was provided.'.format(source) |
||||
self.logger.warning(msg) |
||||
raise TypeError(msg) |
||||
|
||||
self.source = '+{}'.format(result['full']) |
||||
|
||||
if targets: |
||||
# Validate our targerts |
||||
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), |
||||
) |
||||
self.invalid_targets.append(target) |
||||
continue |
||||
|
||||
# store valid phone number |
||||
self.targets.append('+{}'.format(result['full'])) |
||||
else: |
||||
# Send a message to ourselves |
||||
self.targets.append(self.source) |
||||
|
||||
return |
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): |
||||
""" |
||||
Perform Signal API Notification |
||||
""" |
||||
|
||||
if len(self.targets) == 0: |
||||
# There were no services to notify |
||||
self.logger.warning( |
||||
'There were no Signal API 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/json', |
||||
} |
||||
|
||||
# Prepare our payload |
||||
payload = { |
||||
'message': "{}{}".format( |
||||
'' if not self.status else '{} '.format( |
||||
self.asset.ascii(notify_type)), body), |
||||
"number": self.source, |
||||
"recipients": [] |
||||
} |
||||
|
||||
# Determine Authentication |
||||
auth = None |
||||
if self.user: |
||||
auth = (self.user, self.password) |
||||
|
||||
# Set our schema |
||||
schema = 'https' if self.secure else 'http' |
||||
|
||||
# Construct our URL |
||||
notify_url = '%s://%s' % (schema, self.host) |
||||
if isinstance(self.port, int): |
||||
notify_url += ':%d' % self.port |
||||
notify_url += '/v2/send' |
||||
|
||||
# Send in batches if identified to do so |
||||
batch_size = 1 if not self.batch else self.default_batch_size |
||||
|
||||
for index in range(0, len(self.targets), batch_size): |
||||
# Prepare our recipients |
||||
payload['recipients'] = self.targets[index:index + batch_size] |
||||
|
||||
self.logger.debug('Signal API POST URL: %s (cert_verify=%r)' % ( |
||||
notify_url, self.verify_certificate, |
||||
)) |
||||
self.logger.debug('Signal API Payload: %s' % str(payload)) |
||||
|
||||
# Always call throttle before any remote server i/o is made |
||||
self.throttle() |
||||
try: |
||||
r = requests.post( |
||||
notify_url, |
||||
auth=auth, |
||||
data=dumps(payload), |
||||
headers=headers, |
||||
verify=self.verify_certificate, |
||||
timeout=self.request_timeout, |
||||
) |
||||
if r.status_code != requests.codes.ok: |
||||
# We had a problem |
||||
status_str = \ |
||||
NotifySignalAPI.http_response_code_lookup( |
||||
r.status_code) |
||||
|
||||
self.logger.warning( |
||||
'Failed to send {} Signal API notification{}: ' |
||||
'{}{}error={}.'.format( |
||||
len(self.targets[index:index + batch_size]), |
||||
' to {}'.format(self.targets[index]) |
||||
if batch_size == 1 else '(s)', |
||||
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 {} Signal API notification{}.' |
||||
.format( |
||||
len(self.targets[index:index + batch_size]), |
||||
' to {}'.format(self.targets[index]) |
||||
if batch_size == 1 else '(s)', |
||||
)) |
||||
|
||||
except requests.RequestException as e: |
||||
self.logger.warning( |
||||
'A Connection error occured sending {} Signal API ' |
||||
'notification(s).'.format( |
||||
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 = { |
||||
'batch': 'yes' if self.batch else 'no', |
||||
'status': 'yes' if self.status else 'no', |
||||
} |
||||
|
||||
# Extend our parameters |
||||
params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) |
||||
|
||||
# Determine Authentication |
||||
auth = '' |
||||
if self.user and self.password: |
||||
auth = '{user}:{password}@'.format( |
||||
user=NotifySignalAPI.quote(self.user, safe=''), |
||||
password=self.pprint( |
||||
self.password, privacy, mode=PrivacyMode.Secret, safe=''), |
||||
) |
||||
elif self.user: |
||||
auth = '{user}@'.format( |
||||
user=NotifySignalAPI.quote(self.user, safe=''), |
||||
) |
||||
|
||||
default_port = 443 if self.secure else 80 |
||||
|
||||
# So we can strip out our own phone (if present); create a copy of our |
||||
# targets |
||||
if len(self.targets) == 1 and self.source in self.targets: |
||||
targets = [] |
||||
|
||||
elif len(self.targets) == 0: |
||||
# invalid phone-no were specified |
||||
targets = self.invalid_targets |
||||
|
||||
else: |
||||
targets = list(self.targets) |
||||
|
||||
return '{schema}://{auth}{hostname}{port}/{src}/{dst}?{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), |
||||
src=self.source, |
||||
dst='/'.join( |
||||
[NotifySignalAPI.quote(x, safe='') for x in targets]), |
||||
params=NotifySignalAPI.urlencode(params), |
||||
) |
||||
|
||||
@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 entries; split_path() looks after unquoting content for us |
||||
# by default |
||||
results['targets'] = \ |
||||
NotifySignalAPI.split_path(results['fullpath']) |
||||
|
||||
# The hostname is our authentication key |
||||
results['apikey'] = NotifySignalAPI.unquote(results['host']) |
||||
|
||||
if 'from' in results['qsd'] and len(results['qsd']['from']): |
||||
results['source'] = \ |
||||
NotifySignalAPI.unquote(results['qsd']['from']) |
||||
|
||||
elif results['targets']: |
||||
# The from phone no is the first entry in the list otherwise |
||||
results['source'] = results['targets'].pop(0) |
||||
|
||||
# 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'] += \ |
||||
NotifySignalAPI.parse_phone_no(results['qsd']['to']) |
||||
|
||||
# Get Batch Mode Flag |
||||
results['batch'] = \ |
||||
parse_bool(results['qsd'].get('batch', False)) |
||||
|
||||
# Get status switch |
||||
results['status'] = \ |
||||
parse_bool(results['qsd'].get('status', False)) |
||||
|
||||
return results |
||||
@ -0,0 +1,217 @@
|
||||
# -*- coding: utf-8 -*- |
||||
# |
||||
# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com> |
||||
# All rights reserved. |
||||
# |
||||
# This code is licensed under the MIT License. |
||||
# |
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy |
||||
# of this software and associated documentation files(the "Software"), to deal |
||||
# in the Software without restriction, including without limitation the rights |
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell |
||||
# copies of the Software, and to permit persons to whom the Software is |
||||
# furnished to do so, subject to the following conditions : |
||||
# |
||||
# The above copyright notice and this permission notice shall be included in |
||||
# all copies or substantial portions of the Software. |
||||
# |
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR |
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, |
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE |
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
||||
# 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. |
||||
from json import loads |
||||
import mock |
||||
import pytest |
||||
import requests |
||||
from apprise import plugins |
||||
from apprise import Apprise |
||||
from helpers import AppriseURLTester |
||||
|
||||
# Disable logging for a cleaner testing output |
||||
import logging |
||||
logging.disable(logging.CRITICAL) |
||||
|
||||
# Our Testing URLs |
||||
apprise_url_tests = ( |
||||
('signal://', { |
||||
# No host specified |
||||
'instance': TypeError, |
||||
}), |
||||
('signal://:@/', { |
||||
# invalid host |
||||
'instance': TypeError, |
||||
}), |
||||
('signal://localhost', { |
||||
# Just a host provided |
||||
'instance': TypeError, |
||||
}), |
||||
('signal://localhost', { |
||||
# key and secret provided and from but invalid from no |
||||
'instance': TypeError, |
||||
}), |
||||
('signal://localhost/123', { |
||||
# invalid from phone |
||||
'instance': TypeError, |
||||
|
||||
}), |
||||
('signal://localhost/{}/123/'.format('1' * 11), { |
||||
# invalid 'to' phone number |
||||
'instance': plugins.NotifySignalAPI, |
||||
# Notify will fail because it couldn't send to anyone |
||||
'response': False, |
||||
# Our expected url(privacy=True) startswith() response: |
||||
'privacy_url': 'signal://localhost/+{}/123'.format('1' * 11), |
||||
}), |
||||
('signal://localhost:8080/{}/'.format('1' * 11), { |
||||
# one phone number will notify ourselves |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
|
||||
('signal://localhost:8080/?from={}&to={},{}'.format( |
||||
'1' * 11, '2' * 11, '3' * 11), { |
||||
# use get args to acomplish the same thing |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signal://localhost:8080/?from={}&to={},{},{}'.format( |
||||
'1' * 11, '2' * 11, '3' * 11, '5' * 3), { |
||||
# 2 good targets and one invalid one |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signal://localhost:8080/{}/{}/?from={}'.format( |
||||
'1' * 11, '2' * 11, '3' * 11), { |
||||
# If we have from= specified, then all elements take on the to= value |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signals://user@localhost/{}/{}'.format('1' * 11, '3' * 11), { |
||||
# use get args to acomplish the same thing (use source instead of from) |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signals://user:password@localhost/{}/{}'.format('1' * 11, '3' * 11), { |
||||
# use get args to acomplish the same thing (use source instead of from) |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signals://localhost/{}/{}/{}?batch=True'.format( |
||||
'1' * 11, '3' * 11, '4' * 11), { |
||||
# test batch mode |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signals://localhost/{}/{}/{}?status=True'.format( |
||||
'1' * 11, '3' * 11, '4' * 11), { |
||||
# test status switch |
||||
'instance': plugins.NotifySignalAPI, |
||||
}), |
||||
('signal://localhost/{}/{}'.format('1' * 11, '4' * 11), { |
||||
'instance': plugins.NotifySignalAPI, |
||||
# throw a bizzare code forcing us to fail to look it up |
||||
'response': False, |
||||
'requests_response_code': 999, |
||||
}), |
||||
('signal://localhost/{}/{}'.format('1' * 11, '4' * 11), { |
||||
'instance': plugins.NotifySignalAPI, |
||||
# 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_signal_urls(): |
||||
""" |
||||
NotifySignalAPI() Apprise URLs |
||||
|
||||
""" |
||||
|
||||
# Run our general tests |
||||
AppriseURLTester(tests=apprise_url_tests).run_all() |
||||
|
||||
|
||||
@mock.patch('requests.post') |
||||
def test_plugin_signal_edge_cases(mock_post): |
||||
""" |
||||
NotifySignalAPI() 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 |
||||
|
||||
# Initialize some generic (but valid) tokens |
||||
source = '+1 (555) 123-3456' |
||||
target = '+1 (555) 987-5432' |
||||
body = "test body" |
||||
title = "My Title" |
||||
|
||||
# No apikey specified |
||||
with pytest.raises(TypeError): |
||||
plugins.NotifySignalAPI(source=None) |
||||
|
||||
aobj = Apprise() |
||||
assert aobj.add("signals://localhost:231/{}/{}".format(source, target)) |
||||
assert aobj.notify(title=title, body=body) |
||||
|
||||
assert mock_post.call_count == 1 |
||||
|
||||
details = mock_post.call_args_list[0] |
||||
assert details[0][0] == 'https://localhost:231/v2/send' |
||||
payload = loads(details[1]['data']) |
||||
assert payload['message'] == 'My Title\r\ntest body' |
||||
|
||||
# Reset our mock object |
||||
mock_post.reset_mock() |
||||
|
||||
aobj = Apprise() |
||||
assert aobj.add( |
||||
"signals://user@localhost:231/{}/{}?status=True".format( |
||||
source, target)) |
||||
assert aobj.notify(title=title, body=body) |
||||
|
||||
assert mock_post.call_count == 1 |
||||
|
||||
details = mock_post.call_args_list[0] |
||||
assert details[0][0] == 'https://localhost:231/v2/send' |
||||
payload = loads(details[1]['data']) |
||||
# Status flag is set |
||||
assert payload['message'] == '[i] My Title\r\ntest body' |
||||
|
||||
|
||||
@mock.patch('requests.post') |
||||
def test_plugin_signal_test_based_on_feedback(mock_post): |
||||
""" |
||||
NotifySignalAPI() User Feedback Test |
||||
|
||||
""" |
||||
# 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 |
||||
|
||||
body = "test body" |
||||
title = "My Title" |
||||
|
||||
aobj = Apprise() |
||||
aobj.add('signal://10.0.0.112:8080/+12512222222/+12513333333') |
||||
|
||||
assert aobj.notify(title=title, body=body) |
||||
|
||||
assert mock_post.call_count == 1 |
||||
|
||||
details = mock_post.call_args_list[0] |
||||
assert details[0][0] == 'http://10.0.0.112:8080/v2/send' |
||||
payload = loads(details[1]['data']) |
||||
assert payload['message'] == 'My Title\r\ntest body' |
||||
assert payload['number'] == "+12512222222" |
||||
assert payload['recipients'] == ["+12513333333"] |
||||
Loading…
Reference in new issue