extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2025-01-03 20:49:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

SimplePush Support (#146)

Browse Source
This commit is contained in:
Chris Caron 2019-09-07 18:57:02 -04:00 committed by GitHub
parent a420375cc7
commit 24fd7d5baa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 469 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -58,6 +58,7 @@ The table below identifies the services this tool supports and some example serv
| [Rocket.Chat](https://github.com/caronc/apprise/wiki/Notify_rocketchat) | rocket:// or rockets:// | (TCP) 80 or 443 | rocket://user:password@hostname/RoomID/Channel<br />rockets://user:password@hostname:443/#Channel1/#Channel1/RoomID<br />rocket://user:password@hostname/#Channel<br />rocket://webhook@hostname<br />rockets://webhook@hostname/@User/#Channel
| [Ryver](https://github.com/caronc/apprise/wiki/Notify_ryver) | ryver:// | (TCP) 443 | ryver://Organization/Token<br />ryver://botname@Organization/Token
| [SendGrid](https://github.com/caronc/apprise/wiki/Notify_sendgrid) | sendgrid:// | (TCP) 443 | sendgrid://APIToken:FromEmail/<br />sendgrid://APIToken:FromEmail/ToEmail<br />sendgrid://APIToken:FromEmail/ToEmail1/ToEmail2/ToEmailN/
| [SimplePush](https://github.com/caronc/apprise/wiki/Notify_simplepush) | spush:// | (TCP) 443 | spush://apikey<br />spush://salt:password@apikey<br />spush://apikey?event=Apprise
| [Slack](https://github.com/caronc/apprise/wiki/Notify_slack) | slack:// | (TCP) 443 | slack://TokenA/TokenB/TokenC/<br />slack://TokenA/TokenB/TokenC/Channel<br />slack://botname@TokenA/TokenB/TokenC/Channel<br />slack://user@TokenA/TokenB/TokenC/Channel1/Channel2/ChannelN
| [Telegram](https://github.com/caronc/apprise/wiki/Notify_telegram) | tgram:// | (TCP) 443 | tgram://bottoken/ChatID<br />tgram://bottoken/ChatID1/ChatID2/ChatIDN
| [Twitter](https://github.com/caronc/apprise/wiki/Notify_twitter) | twitter:// | (TCP) 443 | twitter://CKey/CSecret/AKey/ASecret<br/>twitter://user@CKey/CSecret/AKey/ASecret<br/>twitter://CKey/CSecret/AKey/ASecret/User1/User2/User2<br/>twitter://CKey/CSecret/AKey/ASecret?mode=tweet

317
apprise/plugins/NotifySimplePush.py Normal file
View File

@ -0,0 +1,317 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 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 os import urandom
from json import loads
import requests
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
# Default our global support flag
CRYPTOGRAPHY_AVAILABLE = False
try:
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher
from cryptography.hazmat.primitives.ciphers import algorithms
from cryptography.hazmat.primitives.ciphers import modes
from cryptography.hazmat.backends import default_backend
from base64 import urlsafe_b64encode
import hashlib
CRYPTOGRAPHY_AVAILABLE = True
except ImportError:
# no problem; this just means the added encryption functionality isn't
# available. You can still send a SimplePush message
pass
class NotifySimplePush(NotifyBase):
"""
A wrapper for SimplePush Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'SimplePush'
# The services URL
service_url = 'https://simplepush.io/'
# The default secure protocol
secure_protocol = 'spush'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_simplepush'
# SimplePush uses the http protocol with SimplePush requests
notify_url = 'https://api.simplepush.io/send'
# The maximum allowable characters allowed in the body per message
body_maxlen = 10000
# Defines the maximum allowable characters in the title
title_maxlen = 1024
# Define object templates
templates = (
'{schema}://{apikey}',
'{schema}://{salt}:{password}@{apikey}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'private': True,
'required': True,
},
# Used for encrypted logins
'password': {
'name': _('Encrypted Password'),
'type': 'string',
'private': True,
},
'salt': {
'name': _('Encrypted Salt'),
'type': 'string',
'private': True,
'map_to': 'user',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'event': {
'name': _('Event'),
'type': 'string',
},
})
def __init__(self, apikey, event=None, **kwargs):
"""
Initialize SimplePush Object
"""
super(NotifySimplePush, self).__init__(**kwargs)
# Store the API key
self.apikey = apikey
# Event Name
self.event = event
# Encrypt Message (providing support is available)
if self.password and self.user and not CRYPTOGRAPHY_AVAILABLE:
# Provide the end user at least some notification that they're
# not getting what they asked for
self.logger.warning(
'SimplePush extended encryption is not supported by this '
'system.')
# Used/cached in _encrypt() function
self._iv = None
self._iv_hex = None
self._key = None
def _encrypt(self, content):
"""
Encrypts message for use with SimplePush
"""
if self._iv is None:
# initialization vector and cache it
self._iv = urandom(algorithms.AES.block_size // 8)
# convert vector into hex string (used in payload)
self._iv_hex = ''.join(["{:02x}".format(ord(self._iv[idx:idx + 1]))
for idx in range(len(self._iv))]).upper()
# encrypted key and cache it
self._key = bytes(bytearray.fromhex(
hashlib.sha1('{}{}'.format(self.password, self.user)
.encode('utf-8')).hexdigest()[0:32]))
padder = padding.PKCS7(algorithms.AES.block_size).padder()
content = padder.update(content.encode()) + padder.finalize()
encryptor = Cipher(
algorithms.AES(self._key),
modes.CBC(self._iv),
default_backend()).encryptor()
return urlsafe_b64encode(
encryptor.update(content) + encryptor.finalize())
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform SimplePush Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-type': "application/x-www-form-urlencoded",
}
# Prepare our payload
payload = {
'key': self.apikey,
}
event = self.event
if self.password and self.user and CRYPTOGRAPHY_AVAILABLE:
body = self._encrypt(body)
title = self._encrypt(title)
payload.update({
'encrypted': 'true',
'iv': self._iv_hex,
})
# prepare SimplePush Object
payload.update({
'msg': body,
'title': title,
})
if event:
payload['event'] = event
self.logger.debug('SimplePush POST URL: %s (cert_verify=%r)' % (
self.notify_url, self.verify_certificate,
))
self.logger.debug('SimplePush Payload: %s' % str(payload))
# We need to rely on the status string returned in the SimplePush
# response
status_str = None
status = None
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.notify_url,
data=payload,
headers=headers,
verify=self.verify_certificate,
)
# Get our SimplePush response (if it's possible)
try:
json_response = loads(r.content)
status_str = json_response.get('message')
status = json_response.get('status')
except (TypeError, ValueError, AttributeError):
# TypeError = r.content is not a String
# ValueError = r.content is Unparsable
# AttributeError = r.content is None
pass
if r.status_code != requests.codes.ok or status != 'OK':
# We had a problem
status_str = status_str if status_str else\
NotifyBase.http_response_code_lookup(r.status_code)
self.logger.warning(
'Failed to send SimplePush notification:'
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug('Response Details:\r\n{}'.format(r.content))
# Return; we're done
return False
else:
self.logger.info('Sent SimplePush notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending SimplePush notification.')
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
return True
def url(self):
"""
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
if self.event:
args['event'] = self.event
# Determine Authentication
auth = ''
if self.user and self.password:
auth = '{salt}:{password}@'.format(
salt=NotifySimplePush.quote(self.user, safe=''),
password=NotifySimplePush.quote(self.password, safe=''),
)
return '{schema}://{auth}{apikey}/?{args}'.format(
schema=self.secure_protocol,
auth=auth,
apikey=NotifySimplePush.quote(self.apikey, safe=''),
args=NotifySimplePush.urlencode(args),
)
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Set the API Key
results['apikey'] = NotifySimplePush.unquote(results['host'])
# Event
if 'event' in results['qsd'] and len(results['qsd']['event']):
# Extract the account sid from an argument
results['event'] = \
NotifySimplePush.unquote(results['qsd']['event'])
return results

1
dev-requirements.txt
View File

@ -5,3 +5,4 @@ pytest
pytest-cov
tox
babel
cryptography

6
packaging/redhat/python-apprise.spec
View File

@ -50,9 +50,9 @@ it easy to access:
Boxcar, ClickSend, Discord, E-Mail, Emby, Faast, Flock, Gitter, Gotify, Growl,
IFTTT, Join, KODI, Mailgun, MatterMost, Matrix, Microsoft Windows Notifications,
Microsoft Teams, MessageBird, MSG91, Nexmo, Notify MyAndroid, Prowl, Pushalot,
PushBullet, Pushjet, Pushover, Rocket.Chat, SendGrid, Slack, Super Toasty,
Stride, Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP,
Webex Teams}
PushBullet, Pushjet, Pushover, Rocket.Chat, SendGrid, SimplePush, Slack,
Super Toasty, Stride, Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC,
XMPP, Webex Teams}
Name: python-%{pypi_name}
Version: 0.7.9

6
setup.py
View File

@ -72,9 +72,9 @@ setup(
keywords='Push Notifications Alerts Email AWS SNS Boxcar ClickSend '
'Discord Dbus Emby Faast Flock Gitter Gnome Gotify Growl IFTTT Join '
'KODI Mailgun Matrix Mattermost MessageBird MSG91 Nexmo Prowl '
'PushBullet Pushjet Pushed Pushover Rocket.Chat Ryver SendGrid Slack '
'Stride Techulus Push Telegram Twilio Twist Twitter XBMC Microsoft '
'MSTeams Windows Webex CLI API',
'PushBullet Pushjet Pushed Pushover Rocket.Chat Ryver SendGrid '
'SimplePush Slack Stride Techulus Push Telegram Twilio Twist Twitter '
'XBMC Microsoft MSTeams Windows Webex CLI API',
author='Chris Caron',
author_email='lead2gold@gmail.com',
packages=find_packages(),

58
test/test_rest_plugins.py
View File

@ -2029,6 +2029,64 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifySimplePush
##################################
('spush://', {
# No api key
'instance': None,
}),
('spush://{}'.format('A' * 14), {
# API Key specified however expected server response
# didn't have 'OK' in JSON response
'instance': plugins.NotifySimplePush,
# Expected notify() response
'notify_response': False,
}),
('spush://{}'.format('X' * 14), {
# API Key valid and expected response was valid
'instance': plugins.NotifySimplePush,
# Set our response to OK
'requests_response_text': {
'status': 'OK',
},
}),
('spush://{}?event=Not%20So%20Good'.format('X' * 14), {
# API Key valid and expected response was valid
'instance': plugins.NotifySimplePush,
# Set our response to something that is not okay
'requests_response_text': {
'status': 'NOT-OK',
},
# Expected notify() response
'notify_response': False,
}),
('spush://salt:pass@{}'.format('X' * 14, 'A' * 16), {
# Now we'll test encrypted messages with new salt
'instance': plugins.NotifySimplePush,
# Set our response to OK
'requests_response_text': {
'status': 'OK',
},
}),
('spush://{}'.format('Y' * 14), {
'instance': plugins.NotifySimplePush,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
# Set a failing message too
'requests_response_text': {
'status': 'BadRequest',
'message': 'Title or message too long',
},
}),
('spush://{}'.format('Z' * 14), {
'instance': plugins.NotifySimplePush,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifySlack
##################################

86
test/test_simplepush_plugin.py Normal file
View File

@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 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 os
import sys
import apprise
try:
# Python v3.4+
from importlib import reload
except ImportError:
try:
# Python v3.0-v3.3
from imp import reload
except ImportError:
# Python v2.7
pass
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
def test_simplepush_plugin(tmpdir):
"""
API: NotifySimplePush Plugin()
"""
suite = tmpdir.mkdir("simplepush")
suite.join("__init__.py").write('')
module_name = 'cryptography'
suite.join("{}.py".format(module_name)).write('raise ImportError()')
# Update our path to point to our new test suite
sys.path.insert(0, str(suite))
for name in list(sys.modules.keys()):
if name.startswith('{}.'.format(module_name)):
del sys.modules[name]
del sys.modules[module_name]
# The following libraries need to be reloaded to prevent
# TypeError: super(type, obj): obj must be an instance or subtype of type
# This is better explained in this StackOverflow post:
# https://stackoverflow.com/questions/31363311/\
# any-way-to-manually-fix-operation-of-\
# super-after-ipython-reload-avoiding-ty
#
reload(sys.modules['apprise.plugins.NotifySimplePush'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])
# imported and therefore the extra encryption offered by SimplePush is
# not available...
obj = apprise.Apprise.instantiate('spush://salt:pass@valid_api_key')
assert obj is not None
# Tidy-up / restore things to how they were
os.unlink(str(suite.join("{}.py".format(module_name))))
reload(sys.modules['apprise.plugins.NotifySimplePush'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])