Re #203 - validate tokens (#204)

* Re #203 - validate tokens

changedetectionio/forms.py

@@ -116,6 +116,20 @@ class ValidateContentFetcherIsReady(object):
                 raise ValidationError(message % (field.data, e))
 
 
+class ValidateTokensList(object):
+    """
+    Validates that a {token} is from a valid set
+    """
+    def __init__(self, message=None):
+        self.message = message
+
+    def __call__(self, form, field):
+        from changedetectionio import notification
+        regex = re.compile('{.*?}')
+        for p in re.findall(regex, field.data):
+            if not p.strip('{}') in notification.valid_tokens:
+                message = field.gettext('Token \'%s\' is not a valid token.')
+                raise ValidationError(message % (p))
 
 class ValidateListRegex(object):
     """
@@ -195,5 +209,5 @@ class globalSettingsForm(Form):
     extract_title_as_title = BooleanField('Extract <title> from document and use as watch title')
     trigger_check = BooleanField('Send test notification on save')
 
-    notification_title = StringField('Notification Title')
-    notification_body = TextAreaField('Notification Body')
+    notification_title = StringField('Notification Title', validators=[validators.Optional(), ValidateTokensList()])
+    notification_body = TextAreaField('Notification Body', validators=[validators.Optional(), ValidateTokensList()])

changedetectionio/notification.py

@@ -1,6 +1,15 @@
 import os
 import apprise
 
+valid_tokens = {
+    'base_url': '',
+    'watch_url': '',
+    'diff_url': '',
+    'preview_url': '',
+    'current_snapshot': ''
+}
+
+
 def process_notification(n_object, datastore):
     apobj = apprise.Apprise()
     for url in n_object['notification_urls']:
@@ -31,7 +40,7 @@ def process_notification(n_object, datastore):
 
 # Notification title + body content parameters get created here.
 def create_notification_parameters(n_object):
-
+    from copy import deepcopy
     # in the case we send a test notification from the main settings, there is no UUID.
     uuid = n_object['uuid'] if 'uuid' in n_object else ''
 
@@ -47,11 +56,19 @@ def create_notification_parameters(n_object):
     diff_url = "{}/diff/{}".format(base_url, uuid)
     preview_url = "{}/preview/{}".format(base_url, uuid)
 
-    return {
+    # Not sure deepcopy is needed here, but why not
+    tokens = deepcopy(valid_tokens)
+
+    # Valid_tokens also used as a field validator
+    tokens.update(
+    {
         'base_url': base_url,
         'watch_url': watch_url,
         'diff_url': diff_url,
         'preview_url': preview_url,
         'current_snapshot': n_object['current_snapshot'] if 'current_snapshot' in n_object else ''
-    }
+    })
+
+    return tokens
+
 

changedetectionio/tests/test_notification.py

@@ -98,10 +98,15 @@ def test_check_notification(client, live_server):
     res = client.post(
         url_for("settings_page"),
         data={"notification_title": "New ChangeDetection.io Notification - {watch_url}",
-              "notification_body": "{base_url}\n{watch_url}\n{preview_url}\n{diff_url}\n{current_snapshot}\n:-)",
+              "notification_body": "BASE URL: {base_url}\n"
+                                   "Watch URL: {watch_url}\n"
+                                   "Preview: {preview_url}\n"
+                                   "Diff URL: {diff_url}\n"
+                                   "Snapshot: {current_snapshot}\n"
+                                   ":-)",
               "notification_urls": "json://foobar.com", #Re #143 should not see that it sent without [test checkbox]
               "minutes_between_check": 180,
-              "fetch_backend": "html_requests",
+              "fetch_backend": "html_requests"
               },
         follow_redirects=True
     )
@@ -145,3 +150,18 @@ def test_check_notification(client, live_server):
     client.get(url_for("api_watch_checknow"), follow_redirects=True)
     time.sleep(3)
     assert os.path.exists("test-datastore/notification.txt") == False
+
+
+    # Now adding a wrong token should give us an error
+    res = client.post(
+        url_for("settings_page"),
+        data={"notification_title": "New ChangeDetection.io Notification - {watch_url}",
+              "notification_body": "Rubbish: {rubbish}\n",
+              "notification_urls": "json://foobar.com",
+              "minutes_between_check": 180,
+              "fetch_backend": "html_requests"
+              },
+        follow_redirects=True
+    )
+
+    assert bytes("is not a valid token".encode('utf-8')) in res.data