From 5ec5414261a4b3a17e993deb8808ea28d8993a7f Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 27 Oct 2019 14:07:47 -0400 Subject: [PATCH] Added support for self-hosted Notica servers (#169) --- apprise/plugins/NotifyNotica.py | 194 +++++++++++++++++++++++++++++--- test/test_rest_plugins.py | 57 +++++++++- 2 files changed, 234 insertions(+), 17 deletions(-) diff --git a/apprise/plugins/NotifyNotica.py b/apprise/plugins/NotifyNotica.py index 0931404f..10debdf7 100644 --- a/apprise/plugins/NotifyNotica.py +++ b/apprise/plugins/NotifyNotica.py @@ -37,14 +37,35 @@ # notica://abc123 # import re +import six import requests from .NotifyBase import NotifyBase +from ..URLBase import PrivacyMode from ..common import NotifyType from ..utils import validate_regex from ..AppriseLocale import gettext_lazy as _ +class NoticaMode(object): + """ + Tracks if we're accessing the notica upstream server or a locally hosted + one. + """ + # We're dealing with a self hosted service + SELFHOSTED = 'selfhosted' + + # We're dealing with the official hosted service at https://notica.us + OFFICIAL = 'official' + + +# Define our Notica Modes +NOTICA_MODES = ( + NoticaMode.SELFHOSTED, + NoticaMode.OFFICIAL, +) + + class NotifyNotica(NotifyBase): """ A wrapper for Notica Notifications @@ -56,8 +77,11 @@ class NotifyNotica(NotifyBase): # The services URL service_url = 'https://notica.us/' + # Insecure protocol (for those self hosted requests) + protocol = 'notica' + # The default protocol (this is secure for notica) - secure_protocol = 'notica' + secure_protocol = 'noticas' # A URL that takes you to the setup/help of the specific protocol setup_url = 'https://github.com/caronc/apprise/wiki/Notify_notica' @@ -71,6 +95,14 @@ class NotifyNotica(NotifyBase): # Define object templates templates = ( '{schema}://{token}', + + # Self-hosted notica servers + '{schema}://{host}/{token}', + '{schema}://{host}:{port}/{token}', + '{schema}://{user}@{host}/{token}', + '{schema}://{user}@{host}:{port}/{token}', + '{schema}://{user}:{password}@{host}/{token}', + '{schema}://{user}:{password}@{host}:{port}/{token}', ) # Define our template tokens @@ -82,9 +114,36 @@ class NotifyNotica(NotifyBase): 'required': True, 'regex': r'^\?*(?P[^/]+)\s*$' }, + 'host': { + 'name': _('Hostname'), + 'type': 'string', + }, + 'port': { + 'name': _('Port'), + 'type': 'int', + 'min': 1, + 'max': 65535, + }, + 'user': { + 'name': _('Username'), + 'type': 'string', + }, + 'password': { + 'name': _('Password'), + 'type': 'string', + 'private': True, + }, }) - def __init__(self, token, **kwargs): + # Define any kwargs we're using + template_kwargs = { + 'headers': { + 'name': _('HTTP Header'), + 'prefix': '+', + }, + } + + def __init__(self, token, headers=None, **kwargs): """ Initialize Notica Object """ @@ -98,6 +157,19 @@ class NotifyNotica(NotifyBase): self.logger.warning(msg) raise TypeError(msg) + # Setup our mode + self.mode = NoticaMode.SELFHOSTED if self.host else NoticaMode.OFFICIAL + + # prepare our fullpath + self.fullpath = kwargs.get('fullpath') + if not isinstance(self.fullpath, six.string_types): + self.fullpath = '/' + + self.headers = {} + if headers: + # Store our extra headers + self.headers.update(headers) + return def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): @@ -113,8 +185,33 @@ class NotifyNotica(NotifyBase): # Prepare our payload payload = 'd:{}'.format(body) - # prepare our notify url - notify_url = self.notify_url.format(token=self.token) + # Auth is used for SELFHOSTED queries + auth = None + + if self.mode is NoticaMode.OFFICIAL: + # prepare our notify url + notify_url = self.notify_url.format(token=self.token) + + else: + # Prepare our self hosted URL + + # Apply any/all header over-rides defined + headers.update(self.headers) + + if self.user: + auth = (self.user, self.password) + + # Set our schema + schema = 'https' if self.secure else 'http' + + # Prepare our notify_url + notify_url = '%s://%s' % (schema, self.host) + if isinstance(self.port, int): + notify_url += ':%d' % self.port + + notify_url += '{fullpath}?token={token}'.format( + fullpath=self.fullpath.strip('/'), + token=self.token) self.logger.debug('Notica POST URL: %s (cert_verify=%r)' % ( notify_url, self.verify_certificate, @@ -129,6 +226,7 @@ class NotifyNotica(NotifyBase): notify_url.format(token=self.token), data=payload, headers=headers, + auth=auth, verify=self.verify_certificate, ) if r.status_code != requests.codes.ok: @@ -174,11 +272,49 @@ class NotifyNotica(NotifyBase): 'verify': 'yes' if self.verify_certificate else 'no', } - return '{schema}://{token}/?{args}'.format( - schema=self.secure_protocol, - token=self.pprint(self.token, privacy, safe=''), - args=NotifyNotica.urlencode(args), - ) + if self.mode == NoticaMode.OFFICIAL: + # Official URLs are easy to assemble + return '{schema}://{token}/?{args}'.format( + schema=self.protocol, + token=self.pprint(self.token, privacy, safe=''), + args=NotifyNotica.urlencode(args), + ) + + # If we reach here then we are assembling a self hosted URL + + # Append our headers into our args + args.update({'+{}'.format(k): v for k, v in self.headers.items()}) + + # Authorization can be used for self-hosted sollutions + auth = '' + + # Determine Authentication + if self.user and self.password: + auth = '{user}:{password}@'.format( + user=NotifyNotica.quote(self.user, safe=''), + password=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe=''), + ) + elif self.user: + auth = '{user}@'.format( + user=NotifyNotica.quote(self.user, safe=''), + ) + + default_port = 443 if self.secure else 80 + + return '{schema}://{auth}{hostname}{port}{fullpath}{token}/?{args}' \ + .format( + schema=self.secure_protocol + if self.secure else self.protocol, + auth=auth, + hostname=NotifyNotica.quote(self.host, safe=''), + port='' if self.port is None or self.port == default_port + else ':{}'.format(self.port), + fullpath=NotifyNotica.quote( + self.fullpath, safe='/'), + token=self.pprint(self.token, privacy, safe=''), + args=NotifyNotica.urlencode(args), + ) @staticmethod def parse_url(url): @@ -192,8 +328,34 @@ class NotifyNotica(NotifyBase): # We're done early as we couldn't load the results return results - # Store our token using the host - results['token'] = NotifyNotica.unquote(results['host']) + # Get unquoted entries + entries = NotifyNotica.split_path(results['fullpath']) + if not entries: + # If there are no path entries, then we're only dealing with the + # official website + results['mode'] = NoticaMode.OFFICIAL + + # Store our token using the host + results['token'] = NotifyNotica.unquote(results['host']) + + # Unset our host + results['host'] = None + + else: + # Otherwise we're running a self hosted instance + results['mode'] = NoticaMode.SELFHOSTED + + # The last element in the list is our token + results['token'] = entries.pop() + + # Re-assemble our full path + results['fullpath'] = \ + '/' if not entries else '/{}/'.format('/'.join(entries)) + + # Add our headers that the user can potentially over-ride if they + # wish to to our returned result set + results['headers'] = results['qsd-'] + results['headers'].update(results['qsd+']) return results @@ -205,12 +367,14 @@ class NotifyNotica(NotifyBase): result = re.match( r'^https?://notica\.us/?' - r'\??(?P[^/&=]+)$', url, re.I) + r'\??(?P[^&]+)([&\s]*(?P.+))?$', url, re.I) if result: return NotifyNotica.parse_url( - '{schema}://{token}'.format( - schema=NotifyNotica.secure_protocol, - token=result.group('token'))) + '{schema}://{token}/{args}'.format( + schema=NotifyNotica.protocol, + token=result.group('token'), + args='' if not result.group('args') + else '?{}'.format(result.group('args')))) return None diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index d4583bab..448871de 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -1415,20 +1415,73 @@ TEST_URLS = ( 'instance': plugins.NotifyNotica, # Our expected url(privacy=True) startswith() response: - 'privacy_url': 'notica://z...z', + 'privacy_url': 'notica://z...z/', + }), + # Native URL with additional arguments + ('https://notica.us/?%s&overflow=upstream' % ('z' * 6), { + 'instance': plugins.NotifyNotica, + + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'notica://z...z/', }), # Token specified ('notica://%s' % ('a' * 6), { 'instance': plugins.NotifyNotica, # Our expected url(privacy=True) startswith() response: - 'privacy_url': 'notica://a...a', + 'privacy_url': 'notica://a...a/', + }), + # Self-Hosted configuration + ('notica://localhost/%s' % ('b' * 6), { + 'instance': plugins.NotifyNotica, + }), + ('notica://user@localhost/%s' % ('c' * 6), { + 'instance': plugins.NotifyNotica, + }), + ('notica://user:pass@localhost/%s/' % ('d' * 6), { + 'instance': plugins.NotifyNotica, + + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'notica://user:****@localhost/d...d', + }), + ('notica://user:pass@localhost/a/path/%s/' % ('r' * 6), { + 'instance': plugins.NotifyNotica, + + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'notica://user:****@localhost/a/path/r...r', + }), + ('notica://localhost:8080/%s' % ('a' * 6), { + 'instance': plugins.NotifyNotica, + }), + ('notica://user:pass@localhost:8080/%s' % ('b' * 6), { + 'instance': plugins.NotifyNotica, + }), + ('noticas://localhost/%s' % ('j' * 6), { + 'instance': plugins.NotifyNotica, + 'privacy_url': 'noticas://localhost/j...j', + }), + ('noticas://user:pass@localhost/%s' % ('e' * 6), { + 'instance': plugins.NotifyNotica, + + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'noticas://user:****@localhost/e...e', + }), + ('noticas://localhost:8080/path/%s' % ('5' * 6), { + 'instance': plugins.NotifyNotica, + 'privacy_url': 'noticas://localhost:8080/path/5...5', + }), + ('noticas://user:pass@localhost:8080/%s' % ('6' * 6), { + 'instance': plugins.NotifyNotica, }), ('notica://%s' % ('b' * 6), { 'instance': plugins.NotifyNotica, # don't include an image by default 'include_image': False, }), + # Test Header overrides + ('notica://localhost:8080//%s/?+HeaderKey=HeaderValue' % ('7' * 6), { + 'instance': plugins.NotifyNotica, + }), ('notica://%s' % ('c' * 6), { 'instance': plugins.NotifyNotica, # force a failure