Re #2742 fixing URL encoding of headers so that '+' in URL is correctly parsed as ' ' (and other url-encodings)

2742-notification-headers-encoding
dgtlmoon 2 months ago
parent 82e0b99b07
commit 1dce9d5549

@ -13,6 +13,7 @@ from loguru import logger
def apprise_custom_api_call_wrapper(body, title, notify_type, *args, **kwargs):
import requests
import json
from urllib.parse import unquote_plus
from apprise.utils import parse_url as apprise_parse_url
from apprise import URLBase
@ -47,7 +48,7 @@ def apprise_custom_api_call_wrapper(body, title, notify_type, *args, **kwargs):
if results:
# Add our headers that the user can potentially over-ride if they wish
# to to our returned result set and tidy entries by unquoting them
headers = {URLBase.unquote(x): URLBase.unquote(y)
headers = {unquote_plus(x): unquote_plus(y)
for x, y in results['qsd+'].items()}
# https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
@ -55,14 +56,14 @@ def apprise_custom_api_call_wrapper(body, title, notify_type, *args, **kwargs):
# but here we are making straight requests, so we need todo convert this against apprise's logic
for k, v in results['qsd'].items():
if not k.strip('+-') in results['qsd+'].keys():
params[URLBase.unquote(k)] = URLBase.unquote(v)
params[unquote_plus(k)] = unquote_plus(v)
# Determine Authentication
auth = ''
if results.get('user') and results.get('password'):
auth = (URLBase.unquote(results.get('user')), URLBase.unquote(results.get('user')))
auth = (unquote_plus(results.get('user')), unquote_plus(results.get('user')))
elif results.get('user'):
auth = (URLBase.unquote(results.get('user')))
auth = (unquote_plus(results.get('user')))
# Try to auto-guess if it's JSON
h = 'application/json; charset=utf-8'

@ -284,7 +284,7 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
# CUSTOM JSON BODY CHECK for POST://
set_original_response()
# https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation
test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123"
test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123&+second=hello+world%20%22space%22"
res = client.post(
url_for("settings_page"),
@ -326,6 +326,7 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
assert j['secret'] == 444
assert j['somebug'] == '网站监测 内容更新了'
# URL check, this will always be converted to lowercase
assert os.path.isfile("test-datastore/notification-url.txt")
with open("test-datastore/notification-url.txt", 'r') as f:
@ -337,6 +338,7 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
with open("test-datastore/notification-headers.txt", 'r') as f:
notification_headers = f.read()
assert 'custom-header: 123' in notification_headers.lower()
assert 'second: hello world "space"' in notification_headers.lower()
# Should always be automatically detected as JSON content type even when we set it as 'Text' (default)

Loading…
Cancel
Save