# -*- 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. import re import requests import hmac from json import dumps from time import time from hashlib import sha1 from itertools import chain try: from urlparse import urlparse except ImportError: from urllib.parse import urlparse from .NotifyBase import NotifyBase from ..URLBase import PrivacyMode from ..utils import parse_bool from ..utils import parse_list from ..utils import validate_regex from ..common import NotifyType from ..common import NotifyImageSize from ..AppriseLocale import gettext_lazy as _ # Default to sending to all devices if nothing is specified DEFAULT_TAG = '@all' # The tags value is an structure containing an array of strings defining the # list of tagged devices that the notification need to be send to, and a # boolean operator (‘and’ / ‘or’) that defines the criteria to match devices # against those tags. IS_TAG = re.compile(r'^[@]?(?P<name>[A-Z0-9]{1,63})$', re.I) # Device tokens are only referenced when developing. # It's not likely you'll send a message directly to a device, but if you do; # this plugin supports it. IS_DEVICETOKEN = re.compile(r'^[A-Z0-9]{64}$', re.I) # Used to break apart list of potential tags by their delimiter into a useable # list. TAGS_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+') class NotifyBoxcar(NotifyBase): """ A wrapper for Boxcar Notifications """ # The default descriptive name associated with the Notification service_name = 'Boxcar' # The services URL service_url = 'https://boxcar.io/' # All boxcar notifications are secure secure_protocol = 'boxcar' # A URL that takes you to the setup/help of the specific protocol setup_url = 'https://github.com/caronc/apprise/wiki/Notify_boxcar' # Boxcar URL notify_url = 'https://boxcar-api.io/api/push/' # Allows the user to specify the NotifyImageSize object image_size = NotifyImageSize.XY_72 # The maximum allowable characters allowed in the body per message body_maxlen = 10000 # Define object templates templates = ( '{schema}://{access_key}/{secret_key}/', '{schema}://{access_key}/{secret_key}/{targets}', ) # Define our template tokens template_tokens = dict(NotifyBase.template_tokens, **{ 'access_key': { 'name': _('Access Key'), 'type': 'string', 'private': True, 'required': True, 'regex': (r'^[A-Z0-9_-]{64}$', 'i'), 'map_to': 'access', }, 'secret_key': { 'name': _('Secret Key'), 'type': 'string', 'private': True, 'required': True, 'regex': (r'^[A-Z0-9_-]{64}$', 'i'), 'map_to': 'secret', }, 'target_tag': { 'name': _('Target Tag ID'), 'type': 'string', 'prefix': '@', 'regex': (r'^[A-Z0-9]{1,63}$', 'i'), 'map_to': 'targets', }, 'target_device': { 'name': _('Target Device ID'), 'type': 'string', 'regex': (r'^[A-Z0-9]{64}$', 'i'), 'map_to': 'targets', }, 'targets': { 'name': _('Targets'), 'type': 'list:string', }, }) # Define our template arguments template_args = dict(NotifyBase.template_args, **{ 'image': { 'name': _('Include Image'), 'type': 'bool', 'default': True, 'map_to': 'include_image', }, 'to': { 'alias_of': 'targets', }, 'access': { 'alias_of': 'access_key', }, 'secret': { 'alias_of': 'secret_key', }, }) def __init__(self, access, secret, targets=None, include_image=True, **kwargs): """ Initialize Boxcar Object """ super().__init__(**kwargs) # Initialize tag list self._tags = list() # Initialize device_token list self.device_tokens = list() # Access Key (associated with project) self.access = validate_regex( access, *self.template_tokens['access_key']['regex']) if not self.access: msg = 'An invalid Boxcar Access Key ' \ '({}) was specified.'.format(access) self.logger.warning(msg) raise TypeError(msg) # Secret Key (associated with project) self.secret = validate_regex( secret, *self.template_tokens['secret_key']['regex']) if not self.secret: msg = 'An invalid Boxcar Secret Key ' \ '({}) was specified.'.format(secret) self.logger.warning(msg) raise TypeError(msg) if not targets: self._tags.append(DEFAULT_TAG) targets = [] # Validate targets and drop bad ones: for target in parse_list(targets): result = IS_TAG.match(target) if result: # store valid tag/alias self._tags.append(result.group('name')) continue result = IS_DEVICETOKEN.match(target) if result: # store valid device self.device_tokens.append(target) continue self.logger.warning( 'Dropped invalid tag/alias/device_token ' '({}) specified.'.format(target), ) # Track whether or not we want to send an image with our notification # or not. self.include_image = include_image return def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): """ Perform Boxcar Notification """ headers = { 'User-Agent': self.app_id, 'Content-Type': 'application/json' } # prepare Boxcar Object payload = { 'aps': { 'badge': 'auto', 'alert': '', }, 'expires': str(int(time() + 30)), } if title: payload['aps']['@title'] = title payload['aps']['alert'] = body if self._tags: payload['tags'] = {'or': self._tags} if self.device_tokens: payload['device_tokens'] = self.device_tokens # Source picture should be <= 450 DP wide, ~2:1 aspect. image_url = None if not self.include_image \ else self.image_url(notify_type) if image_url: # Set our image payload['@img'] = image_url # Acquire our hostname host = urlparse(self.notify_url).hostname # Calculate signature. str_to_sign = "%s\n%s\n%s\n%s" % ( "POST", host, "/api/push", dumps(payload)) h = hmac.new( bytearray(self.secret, 'utf-8'), bytearray(str_to_sign, 'utf-8'), sha1, ) params = NotifyBoxcar.urlencode({ "publishkey": self.access, "signature": h.hexdigest(), }) notify_url = '%s?%s' % (self.notify_url, params) self.logger.debug('Boxcar POST URL: %s (cert_verify=%r)' % ( notify_url, self.verify_certificate, )) self.logger.debug('Boxcar Payload: %s' % str(payload)) # Always call throttle before any remote server i/o is made self.throttle() try: r = requests.post( notify_url, data=dumps(payload), headers=headers, verify=self.verify_certificate, timeout=self.request_timeout, ) # Boxcar returns 201 (Created) when successful if r.status_code != requests.codes.created: # We had a problem status_str = \ NotifyBoxcar.http_response_code_lookup(r.status_code) self.logger.warning( 'Failed to send Boxcar 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 Boxcar notification.') except requests.RequestException as e: self.logger.warning( 'A Connection error occurred sending Boxcar ' 'notification to %s.' % (host)) self.logger.debug('Socket Exception: %s' % str(e)) # Return; we're done return False return True def url(self, privacy=False, *args, **kwargs): """ Returns the URL built dynamically based on specified arguments. """ # Define any URL parameters params = { 'image': 'yes' if self.include_image else 'no', } # Extend our parameters params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) return '{schema}://{access}/{secret}/{targets}?{params}'.format( schema=self.secure_protocol, access=self.pprint(self.access, privacy, safe=''), secret=self.pprint( self.secret, privacy, mode=PrivacyMode.Secret, safe=''), targets='/'.join([ NotifyBoxcar.quote(x, safe='') for x in chain( self._tags, self.device_tokens) if x != DEFAULT_TAG]), params=NotifyBoxcar.urlencode(params), ) def __len__(self): """ Returns the number of targets associated with this notification """ targets = len(self._tags) + len(self.device_tokens) # DEFAULT_TAG is set if no tokens/tags are otherwise set return targets if targets > 0 else 1 @staticmethod def parse_url(url): """ Parses the URL and returns it broken apart into a dictionary. """ results = NotifyBase.parse_url(url, verify_host=False) if not results: # We're done early return None # The first token is stored in the hostname results['access'] = NotifyBoxcar.unquote(results['host']) # Get our entries; split_path() looks after unquoting content for us # by default entries = NotifyBoxcar.split_path(results['fullpath']) # Now fetch the remaining tokens results['secret'] = entries.pop(0) if entries else None # Our recipients make up the remaining entries of our array results['targets'] = entries # The 'to' makes it easier to use yaml configuration if 'to' in results['qsd'] and len(results['qsd']['to']): results['targets'] += \ NotifyBoxcar.parse_list(results['qsd'].get('to')) # Access if 'access' in results['qsd'] and results['qsd']['access']: results['access'] = NotifyBoxcar.unquote( results['qsd']['access'].strip()) # Secret if 'secret' in results['qsd'] and results['qsd']['secret']: results['secret'] = NotifyBoxcar.unquote( results['qsd']['secret'].strip()) # Include images with our message results['include_image'] = \ parse_bool(results['qsd'].get('image', True)) return results