#1170 allow json data in OneSignal template arguments (#1171)

Co-authored-by: phantom <phantom@sictamil.com>
This commit is contained in:
phantom943 2024-07-23 00:26:28 +02:00 committed by GitHub
parent 1e4b4355ce
commit 9620901afc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 122 additions and 3 deletions

View File

@ -40,6 +40,8 @@ from itertools import chain
from .base import NotifyBase from .base import NotifyBase
from ..common import NotifyType from ..common import NotifyType
from ..common import NotifyImageSize from ..common import NotifyImageSize
from ..utils import decode_b64_dict
from ..utils import encode_b64_dict
from ..utils import validate_regex from ..utils import validate_regex
from ..utils import parse_list from ..utils import parse_list
from ..utils import parse_bool from ..utils import parse_bool
@ -167,6 +169,12 @@ class NotifyOneSignal(NotifyBase):
'default': True, 'default': True,
'map_to': 'use_contents', 'map_to': 'use_contents',
}, },
'decode': {
'name': _('Decode Template Args'),
'type': 'bool',
'default': False,
'map_to': 'decode_tpl_args',
},
'template': { 'template': {
'alias_of': 'template', 'alias_of': 'template',
}, },
@ -195,7 +203,8 @@ class NotifyOneSignal(NotifyBase):
def __init__(self, app, apikey, targets=None, include_image=True, def __init__(self, app, apikey, targets=None, include_image=True,
template=None, subtitle=None, language=None, batch=None, template=None, subtitle=None, language=None, batch=None,
use_contents=None, custom=None, postback=None, **kwargs): use_contents=None, decode_tpl_args=None,
custom=None, postback=None, **kwargs):
""" """
Initialize OneSignal Initialize OneSignal
@ -228,6 +237,11 @@ class NotifyOneSignal(NotifyBase):
use_contents if use_contents is not None else use_contents if use_contents is not None else
self.template_args['contents']['default']) else False self.template_args['contents']['default']) else False
# Prepare Decode Template Arguments Flag
self.decode_tpl_args = True if (
decode_tpl_args if decode_tpl_args is not None else
self.template_args['decode']['default']) else False
# Place a thumbnail image inline with the message body # Place a thumbnail image inline with the message body
self.include_image = include_image self.include_image = include_image
@ -301,6 +315,9 @@ class NotifyOneSignal(NotifyBase):
# Custom Data # Custom Data
self.custom_data = {} self.custom_data = {}
if custom and isinstance(custom, dict): if custom and isinstance(custom, dict):
if self.decode_tpl_args:
custom = decode_b64_dict(custom)
self.custom_data.update(custom) self.custom_data.update(custom)
elif custom: elif custom:
@ -471,9 +488,12 @@ class NotifyOneSignal(NotifyBase):
# Extend our parameters # Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
custom_data, needs_decoding = encode_b64_dict(self.custom_data)
# custom_data, needs_decoding = self.custom_data, False
# Save our template data # Save our template data
params.update( params.update(
{':{}'.format(k): v for k, v in self.custom_data.items()}) {':{}'.format(k): v for k, v in custom_data.items()}
)
# Save our postback data # Save our postback data
params.update( params.update(
@ -482,6 +502,11 @@ class NotifyOneSignal(NotifyBase):
if self.use_contents != self.template_args['contents']['default']: if self.use_contents != self.template_args['contents']['default']:
params['contents'] = 'yes' if self.use_contents else 'no' params['contents'] = 'yes' if self.use_contents else 'no'
if (self.decode_tpl_args != self.template_args['decode']['default']
or needs_decoding):
params['decode'] = 'yes' if (self.decode_tpl_args or
needs_decoding) else 'no'
return '{schema}://{tp_id}{app}@{apikey}/{targets}?{params}'.format( return '{schema}://{tp_id}{app}@{apikey}/{targets}?{params}'.format(
schema=self.secure_protocol, schema=self.secure_protocol,
tp_id='{}:'.format( tp_id='{}:'.format(
@ -568,6 +593,13 @@ class NotifyOneSignal(NotifyBase):
'contents', 'contents',
NotifyOneSignal.template_args['contents']['default'])) NotifyOneSignal.template_args['contents']['default']))
# Get Use Contents Boolean (if set)
results['decode_tpl_args'] = \
parse_bool(
results['qsd'].get(
'decode',
NotifyOneSignal.template_args['decode']['default']))
# The API Key is stored in the hostname # The API Key is stored in the hostname
results['apikey'] = NotifyOneSignal.unquote(results['host']) results['apikey'] = NotifyOneSignal.unquote(results['host'])

View File

@ -25,13 +25,15 @@
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # 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 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE. # POSSIBILITY OF SUCH DAMAGE.
import copy
import re import re
import sys import sys
import json import json
import contextlib import contextlib
import os import os
import locale import locale
import typing
import base64
from itertools import chain from itertools import chain
from os.path import expanduser from os.path import expanduser
from functools import reduce from functools import reduce
@ -1600,3 +1602,35 @@ def dict_full_update(dict1, dict2):
_merge(dict1, dict2) _merge(dict1, dict2)
return return
def decode_b64_dict(di: dict) -> dict:
di = copy.deepcopy(di)
for k, v in di.items():
if not isinstance(v, str) or not v.startswith("b64:"):
continue
try:
parsed_v = base64.b64decode(v[4:])
parsed_v = json.loads(parsed_v)
except Exception:
parsed_v = v
di[k] = parsed_v
return di
def encode_b64_dict(
di: dict
) -> typing.Tuple[dict, bool]:
di = copy.deepcopy(di)
needs_decoding = False
for k, v in di.items():
if isinstance(v, str):
continue
try:
encoded = base64.urlsafe_b64encode(json.dumps(v).encode())
encoded = "b64:{}".format(encoded.decode())
needs_decoding = True
except Exception:
encoded = str(v)
di[k] = encoded
return di, needs_decoding

View File

@ -361,3 +361,56 @@ def test_plugin_onesignal_notifications(mock_post):
'small_icon': 'https://github.com/caronc/apprise' 'small_icon': 'https://github.com/caronc/apprise'
'/raw/master/apprise/assets/themes/default/apprise-info-32x32.png', '/raw/master/apprise/assets/themes/default/apprise-info-32x32.png',
'include_external_user_ids': ['@user']} 'include_external_user_ids': ['@user']}
# Test without decoding parameters
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=b64:eyJhIjoxLCJiIjoyfQ==&decode=no')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {"par": "b64:eyJhIjoxLCJiIjoyfQ=="}
# Now same with loading parameters
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=b64:eyJhIjoxLCJiIjoyfQ==&decode=yes')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {"par": {"a": 1, "b": 2}}
# Test bad data in general
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=garbage1&decode=yes')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {"par": 'garbage1'}
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=b64:garbage2&decode=yes')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {"par": 'b64:garbage2'}
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=b64:garbage3==&decode=yes')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {"par": 'b64:garbage3=='}
# Now same with not-base64 parameters
instance = Apprise.instantiate(
'onesignal://templateid:appid@apikey/@user/'
'?:par=eyJhIjoxLCJiIjoyfQ==&:par2=123&decode=yes')
assert isinstance(instance, NotifyOneSignal) and \
instance.custom_data == {
"par": "eyJhIjoxLCJiIjoyfQ==", "par2": "123"
}
# Test incorrect base64 parameters. Second one has incorrect padding
url = 'onesignal://templateid:appid@apikey/@user/' \
'?:par=b64:1234=&:par2=b64:eyJhIjoxLCJiIjoyfQ&' \
':par3=b64:eyJhIjoxLCJiIjoyfQ==&decode=yes'
instance = Apprise.instantiate(url)
assert isinstance(instance, NotifyOneSignal) and instance.custom_data == {
"par": "b64:1234=",
"par2": "b64:eyJhIjoxLCJiIjoyfQ",
"par3": {"a": 1, "b": 2}
}