Notification - Make sure all notification tokens have something set even for form validation, fixes hassio:// with {{ watch_uuid }} in notification URL form (#3504)

changedetectionio/blueprint/ui/notification.py

@@ -2,6 +2,7 @@ from flask import Blueprint, request, make_response
 import random
 from loguru import logger
 
+from changedetectionio.notification_service import NotificationContextData
 from changedetectionio.store import ChangeDetectionStore
 from changedetectionio.auth_decorator import login_optionally_required
 
@@ -19,6 +20,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
         import apprise
         from changedetectionio.notification.handler import process_notification
         from changedetectionio.notification.apprise_plugin.assets import apprise_asset
+        from changedetectionio.jinja2_custom import render as jinja_render
 
         from changedetectionio.notification.apprise_plugin.custom_handlers import apprise_http_custom_handler
 
@@ -61,16 +63,20 @@ def construct_blueprint(datastore: ChangeDetectionStore):
             return 'Error: No Notification URLs set/found'
 
         for n_url in notification_urls:
+            # We are ONLY validating the apprise:// part here, convert all tags to something so as not to break apprise URLs
+            generic_notification_context_data = NotificationContextData()
+            generic_notification_context_data.set_random_for_validation()
+            n_url = jinja_render(template_str=n_url, **generic_notification_context_data).strip()
             if len(n_url.strip()):
                 if not apobj.add(n_url):
                     return f'Error:  {n_url} is not a valid AppRise URL.'
 
         try:
             # use the same as when it is triggered, but then override it with the form test values
-            n_object = {
+            n_object = NotificationContextData({
                 'watch_url': request.form.get('window_url', "https://changedetection.io"),
                 'notification_urls': notification_urls
-            }
+            })
 
             # Only use if present, if not set in n_object it should use the default system value
             if 'notification_format' in request.form and request.form['notification_format'].strip():

changedetectionio/forms.py

@@ -5,6 +5,7 @@ from wtforms.widgets.core import TimeInput
 
 from changedetectionio.blueprint.rss import RSS_FORMAT_TYPES
 from changedetectionio.conditions.form import ConditionFormRow
+from changedetectionio.notification_service import NotificationContextData
 from changedetectionio.strtobool import strtobool
 
 from wtforms import (
@@ -469,11 +470,16 @@ class ValidateAppRiseServers(object):
         import apprise
         from .notification.apprise_plugin.assets import apprise_asset
         from .notification.apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+        from changedetectionio.jinja2_custom import render as jinja_render
 
         apobj = apprise.Apprise(asset=apprise_asset)
 
         for server_url in field.data:
-            url = server_url.strip()
+            generic_notification_context_data = NotificationContextData()
+            # Make sure something is atleast in all those regular token fields
+            generic_notification_context_data.set_random_for_validation()
+
+            url = jinja_render(template_str=server_url.strip(), **generic_notification_context_data).strip()
             if url.startswith("#"):
                 continue
 
@@ -500,7 +506,7 @@ class ValidateJinja2Template(object):
             jinja2_env = create_jinja_env(loader=BaseLoader)
 
             # Add notification tokens for validation
-            jinja2_env.globals.update(notification.valid_tokens)
+            jinja2_env.globals.update(NotificationContextData())
             if hasattr(field, 'extra_notification_tokens'):
                 jinja2_env.globals.update(field.extra_notification_tokens)
 

changedetectionio/notification/__init__.py

@@ -16,20 +16,3 @@ valid_notification_formats = {
     default_notification_format_for_watch: default_notification_format_for_watch
 }
 
-
-valid_tokens = {
-    'base_url': '',
-    'current_snapshot': '',
-    'diff': '',
-    'diff_added': '',
-    'diff_full': '',
-    'diff_patch': '',
-    'diff_removed': '',
-    'diff_url': '',
-    'preview_url': '',
-    'triggered_text': '',
-    'watch_tag': '',
-    'watch_title': '',
-    'watch_url': '',
-    'watch_uuid': '',
-}

changedetectionio/notification/handler.py

@@ -3,16 +3,22 @@ import time
 import apprise
 from loguru import logger
 from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
+from ..notification_service import NotificationContextData
 
-def process_notification(n_object, datastore):
+
+def process_notification(n_object: NotificationContextData, datastore):
     from changedetectionio.jinja2_custom import render as jinja_render
     from . import default_notification_format_for_watch, default_notification_format, valid_notification_formats
     # be sure its registered
     from .apprise_plugin.custom_handlers import apprise_http_custom_handler
 
+    if not isinstance(n_object, NotificationContextData):
+        raise TypeError(f"Expected NotificationContextData, got {type(n_object)}")
+
     now = time.time()
     if n_object.get('notification_timestamp'):
         logger.trace(f"Time since queued {now-n_object['notification_timestamp']:.3f}s")
+
     # Insert variables into the notification content
     notification_parameters = create_notification_parameters(n_object, datastore)
 
@@ -141,17 +147,15 @@ def process_notification(n_object, datastore):
 
 # Notification title + body content parameters get created here.
 # ( Where we prepare the tokens in the notification to be replaced with actual values )
-def create_notification_parameters(n_object, datastore):
+def create_notification_parameters(n_object: NotificationContextData, datastore):
-    from copy import deepcopy
+    if not isinstance(n_object, NotificationContextData):
-    from . import valid_tokens
+        raise TypeError(f"Expected NotificationContextData, got {type(n_object)}")
 
-    # in the case we send a test notification from the main settings, there is no UUID.
+    watch = datastore.data['watching'].get(n_object['uuid'])
-    uuid = n_object['uuid'] if 'uuid' in n_object else ''
+    if watch:
-
+        watch_title = datastore.data['watching'][n_object['uuid']].label
-    if uuid:
-        watch_title = datastore.data['watching'][uuid].label
         tag_list = []
-        tags = datastore.get_all_tags_for_watch(uuid)
+        tags = datastore.get_all_tags_for_watch(n_object['uuid'])
         if tags:
             for tag_uuid, tag in tags.items():
                 tag_list.append(tag.get('title'))
@@ -166,14 +170,10 @@ def create_notification_parameters(n_object, datastore):
 
     watch_url = n_object['watch_url']
 
-    diff_url = "{}/diff/{}".format(base_url, uuid)
+    diff_url = "{}/diff/{}".format(base_url, n_object['uuid'])
-    preview_url = "{}/preview/{}".format(base_url, uuid)
+    preview_url = "{}/preview/{}".format(base_url, n_object['uuid'])
 
-    # Not sure deepcopy is needed here, but why not
+    n_object.update(
-    tokens = deepcopy(valid_tokens)
-
-    # Valid_tokens also used as a field validator
-    tokens.update(
         {
             'base_url': base_url,
             'diff_url': diff_url,
@@ -181,13 +181,10 @@ def create_notification_parameters(n_object, datastore):
             'watch_tag': watch_tag if watch_tag is not None else '',
             'watch_title': watch_title if watch_title is not None else '',
             'watch_url': watch_url,
-            'watch_uuid': uuid,
+            'watch_uuid': n_object['uuid'],
         })
 
-    # n_object will contain diff, diff_added etc etc
+    if watch:
-    tokens.update(n_object)
+        n_object.update(datastore.data['watching'].get(n_object['uuid']).extra_notification_token_values())
 
-    if uuid:
+    return n_object
-        tokens.update(datastore.data['watching'].get(uuid).extra_notification_token_values())
-
-    return tokens

changedetectionio/notification_service.py

@@ -6,9 +6,48 @@ Extracted from update_worker.py to provide standalone notification functionality
 for both sync and async workers
 """
 
-import time
 from loguru import logger
+import time
 
+# What is passed around as notification context, also used as the complete list of valid {{ tokens }}
+class NotificationContextData(dict):
+    def __init__(self, initial_data=None, **kwargs):
+        super().__init__({
+            'current_snapshot': None,
+            'diff': None,
+            'diff_added': None,
+            'diff_full': None,
+            'diff_patch': None,
+            'diff_removed': None,
+            'notification_timestamp': time.time(),
+            'screenshot': None,
+            'triggered_text': None,
+            'uuid': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX',  # Converted to 'watch_uuid' in create_notification_parameters
+            'watch_url': 'https://WATCH-PLACE-HOLDER/',
+            'base_url': None,
+            'diff_url': None,
+            'preview_url': None,
+            'watch_tag': None,
+            'watch_title': None
+        })
+
+        # Apply any initial data passed in
+        self.update({'watch_uuid': self.get('uuid')})
+        if initial_data:
+            self.update(initial_data)
+
+        # Apply any keyword arguments
+        if kwargs:
+            self.update(kwargs)
+
+    def set_random_for_validation(self):
+        import random, string
+        """Randomly fills all dict keys with random strings (for validation/testing)."""
+        for key in self.keys():
+            if key in ['uuid', 'time', 'watch_uuid']:
+                continue
+            rand_str = 'RANDOM-PLACEHOLDER-'+''.join(random.choices(string.ascii_letters + string.digits, k=12))
+            self[key] = rand_str
 
 class NotificationService:
     """
@@ -20,13 +59,16 @@ class NotificationService:
         self.datastore = datastore
         self.notification_q = notification_q
     
-    def queue_notification_for_watch(self, n_object, watch):
+    def queue_notification_for_watch(self, n_object: NotificationContextData, watch):
         """
         Queue a notification for a watch with full diff rendering and template variables
         """
         from changedetectionio import diff
         from changedetectionio.notification import default_notification_format_for_watch
 
+        if not isinstance(n_object, NotificationContextData):
+            raise TypeError(f"Expected NotificationContextData, got {type(n_object)}")
+
         dates = []
         trigger_text = ''
 
@@ -83,11 +125,11 @@ class NotificationService:
             'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True, line_feed_sep=line_feed_sep, html_colour=html_colour_enable),
             'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, patch_format=True),
             'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, line_feed_sep=line_feed_sep),
-            'notification_timestamp': now,
             'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
             'triggered_text': triggered_text,
             'uuid': watch.get('uuid') if watch else None,
             'watch_url': watch.get('url') if watch else None,
+            'watch_uuid': watch.get('uuid') if watch else None,
         })
 
         if watch:
@@ -140,7 +182,7 @@ class NotificationService:
         """
         Send notification when content changes are detected
         """
-        n_object = {}
+        n_object = NotificationContextData()
         watch = self.datastore.data['watching'].get(watch_uuid)
         if not watch:
             return
@@ -183,11 +225,13 @@ class NotificationService:
         if not watch:
             return
 
-        n_object = {'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
+        n_object = NotificationContextData({
-                    'notification_body': "Your configured CSS/xPath filters of '{}' for {{{{watch_url}}}} did not appear on the page after {} attempts, did the page change layout?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\nThanks - Your omniscient changedetection.io installation :)\n".format(
+            'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
-                        ", ".join(watch['include_filters']),
+            'notification_body': "Your configured CSS/xPath filters of '{}' for {{{{watch_url}}}} did not appear on the page after {} attempts, did the page change layout?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\nThanks - Your omniscient changedetection.io installation :)\n".format(
-                        threshold),
+                ", ".join(watch['include_filters']),
-                    'notification_format': 'text'}
+                threshold),
+            'notification_format': 'text'
+        })
 
         if len(watch['notification_urls']):
             n_object['notification_urls'] = watch['notification_urls']
@@ -215,12 +259,14 @@ class NotificationService:
         if not watch:
             return
         threshold = self.datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts')
-        n_object = {'notification_title': "Changedetection.io - Alert - Browser step at position {} could not be run".format(step_n+1),
+        n_object = NotificationContextData({
-                    'notification_body': "Your configured browser step at position {} for {{{{watch_url}}}} "
+            'notification_title': "Changedetection.io - Alert - Browser step at position {} could not be run".format(step_n+1),
-                                         "did not appear on the page after {} attempts, did the page change layout? "
+            'notification_body': "Your configured browser step at position {} for {{{{watch_url}}}} "
-                                         "Does it need a delay added?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\n"
+                                 "did not appear on the page after {} attempts, did the page change layout? "
-                                         "Thanks - Your omniscient changedetection.io installation :)\n".format(step_n+1, threshold),
+                                 "Does it need a delay added?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\n"
-                    'notification_format': 'text'}
+                                 "Thanks - Your omniscient changedetection.io installation :)\n".format(step_n+1, threshold),
+            'notification_format': 'text'
+        })
 
         if len(watch['notification_urls']):
             n_object['notification_urls'] = watch['notification_urls']

changedetectionio/tests/test_notification.py

@@ -284,6 +284,27 @@ def test_notification_validation(client, live_server, measure_memory_usage):
     )
 
 
+def test_notification_urls_jinja2_apprise_integration(client, live_server, measure_memory_usage):
+
+    #
+    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation
+    test_notification_url = "hassio://127.0.0.1/longaccesstoken?verify=no&nid={{watch_uuid}}"
+
+    res = client.post(
+        url_for("settings.settings_page"),
+        data={
+              "application-fetch_backend": "html_requests",
+              "application-minutes_between_check": 180,
+              "application-notification_body": '{ "url" : "{{ watch_url }}", "secret": 444, "somebug": "网站监测 内容更新了" }',
+              "application-notification_format": default_notification_format,
+              "application-notification_urls": test_notification_url,
+              # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
+              "application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }} ",
+              },
+        follow_redirects=True
+    )
+    assert b'Settings updated' in res.data
+
 
 def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_memory_usage):
     
@@ -294,7 +315,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://')+"?status_code=204&xxx={{ watch_url }}&+custom-header=123&+second=hello+world%20%22space%22"
+    test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?status_code=204&watch_uuid={{ watch_uuid }}&xxx={{ watch_url }}&now={% now 'Europe/London', '%Y-%m-%d' %}&+custom-header=123&+second=hello+world%20%22space%22"
 
     res = client.post(
         url_for("settings.settings_page"),
@@ -320,6 +341,7 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
     )
 
     assert b"Watch added" in res.data
+    watch_uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))
 
     wait_for_all_checks(client)
     set_modified_response()
@@ -349,6 +371,11 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
         assert 'xxx=http' in notification_url
         # apprise style headers should be stripped
         assert 'custom-header' not in notification_url
+        # Check jinja2 custom arrow/jinja2-time replace worked
+        assert 'now=2' in notification_url
+        # Check our watch_uuid appeared
+        assert f'watch_uuid={watch_uuid}' in notification_url
+
 
     with open("test-datastore/notification-headers.txt", 'r') as f:
         notification_headers = f.read()
@@ -416,7 +443,6 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
     assert res.status_code != 400
     assert res.status_code != 500
 
-
     with open("test-datastore/notification.txt", 'r') as f:
         x = f.read()
         assert test_body in x