lee/changedetection.io
1
0
Fork 0
mirror of https://github.com/dgtlmoon/changedetection.io.git synced 2026-09-24 22:36:33 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Re #2866 - Make sure any HTML type notifications have their content escaped - except for our added/remove/changed markup

Browse Source
This commit is contained in:
dgtlmoon
2025-01-09 22:43:52 +01:00
parent 5a0ef8fc01
commit 52926efbd1
5 changed files with 125 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changedetectionio/diff.py
+4 -4
View File
@@ -43,19 +43,19 @@ def customSequenceMatcher(
yield before[alo:ahi]
elif include_removed and tag == 'delete':
if html_colour:
yield [f'<span style="{REMOVED_STYLE}">{line}</span>' for line in same_slicer(before, alo, ahi)]
yield [f'<span class="cdio" style="{REMOVED_STYLE}">{line}</span>' for line in same_slicer(before, alo, ahi)]
else:
yield [f"(removed) {line}" for line in same_slicer(before, alo, ahi)] if include_change_type_prefix else same_slicer(before, alo, ahi)
elif include_replaced and tag == 'replace':
if html_colour:
yield [f'<span style="{REMOVED_STYLE}">{line}</span>' for line in same_slicer(before, alo, ahi)] + \
[f'<span style="{ADDED_STYLE}">{line}</span>' for line in same_slicer(after, blo, bhi)]
yield [f'<span class="cdio" style="{REMOVED_STYLE}">{line}</span>' for line in same_slicer(before, alo, ahi)] + \
[f'<span class="cdio" style="{ADDED_STYLE}">{line}</span>' for line in same_slicer(after, blo, bhi)]
else:
yield [f"(changed) {line}" for line in same_slicer(before, alo, ahi)] + \
[f"(into) {line}" for line in same_slicer(after, blo, bhi)] if include_change_type_prefix else same_slicer(before, alo, ahi) + same_slicer(after, blo, bhi)
elif include_added and tag == 'insert':
if html_colour:
yield [f'<span style="{ADDED_STYLE}">{line}</span>' for line in same_slicer(after, blo, bhi)]
yield [f'<span class="cdio" style="{ADDED_STYLE}">{line}</span>' for line in same_slicer(after, blo, bhi)]
else:
yield [f"(added) {line}" for line in same_slicer(after, blo, bhi)] if include_change_type_prefix else same_slicer(after, blo, bhi)
changedetectionio/flask_app.py
+9 -2
View File
@@ -12,6 +12,7 @@ import threading
import time
import timeago
from .html_tools import escape_mixed_content
from .processors import find_processors, get_parent_module, get_custom_watch_obj_for_processor
from .safe_jinja import render as jinja_render
from changedetectionio.strtobool import strtobool
@@ -539,6 +540,7 @@ def changedetection_app(config=None, datastore_o=None):
import apprise
import random
from .apprise_asset import asset
from .update_worker import build_notification_object_for_watch
apobj = apprise.Apprise(asset=asset)
# so that the custom endpoints are registered
@@ -610,9 +612,14 @@ def changedetection_app(config=None, datastore_o=None):
else:
n_object['notification_body'] = "Test body"
n_object['as_async'] = False
n_object.update(watch.extra_notification_token_values())
n_object = build_notification_object_for_watch(watch, n_object, datastore.data['settings']['application'].get('notification_body'))
if n_object['notification_format'].startswith('HTML'):
n_object['notification_body'] = escape_mixed_content(n_object['notification_body'])
from .notification import process_notification
n_object['as_async'] = False
# Now we send the notification_body after everything is compiled
sent_obj = process_notification(n_object, datastore)
except Exception as e:
changedetectionio/html_tools.py
+37
View File
@@ -500,3 +500,40 @@ def get_triggered_text(content, trigger_text):
i += 1
return triggered_text
from bs4 import BeautifulSoup
import html
def escape_mixed_content(document):
import uuid
# Parse the document as HTML
# Generate a single random hash for placeholders
random_hash = f"__PLACEHOLDER_{uuid.uuid4().hex}__"
placeholder_map = []
# <br> to something else so we can preserve them
random_hash_br = f"__BR_{uuid.uuid4().hex}__"
document = document.replace('<br>', random_hash_br)
soup = BeautifulSoup(document, 'html.parser')
# Find all <span class="cdio"> and <br>/<br/>
for tag in soup.find_all("span", class_="cdio"):
placeholder_map.append(str(tag)) # Save the tag as a string
tag.replace_with(random_hash) # Replace tag with the placeholder
# Escape the entire document
escaped_html = html.escape(str(soup))
# Restore all occurrences of placeholders with the original tags
for original_tag in placeholder_map:
escaped_html = escaped_html.replace(random_hash, original_tag, 1) # Replace one occurrence at a time
escaped_html = escaped_html.replace( random_hash_br, "<br>")
return escaped_html
changedetectionio/notification.py
+3 -1
View File
@@ -4,6 +4,7 @@ from apprise import NotifyFormat
import apprise
from loguru import logger
from changedetectionio.html_tools import escape_mixed_content
valid_tokens = {
'base_url': '',
@@ -85,6 +86,8 @@ def process_notification(n_object, datastore):
n_body = n_body.replace("\n", '<br>')
n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)
if n_object['notification_format'].startswith('HTML'):
n_body = escape_mixed_content(n_body)
url = url.strip()
if url.startswith('#'):
@@ -161,7 +164,6 @@ def process_notification(n_object, datastore):
attach=n_object.get('screenshot', None)
)
# Returns empty string if nothing found, multi-line string otherwise
log_value = logs.getvalue()
changedetectionio/update_worker.py
+72 -68
View File
@@ -16,6 +16,77 @@ import time
from loguru import logger
def build_notification_object_for_watch(watch, n_object, default_app_settings_notification_format):
from changedetectionio import diff
from changedetectionio.notification import default_notification_format_for_watch
dates = []
trigger_text = ''
if watch:
watch_history = watch.history
dates = list(watch_history.keys())
trigger_text = watch.get('trigger_text', [])
# Add text that was triggered
if len(dates):
snapshot_contents = watch.get_history_snapshot(dates[-1])
else:
snapshot_contents = "No snapshot/history available, the watch should fetch atleast once."
# If we ended up here with "System default"
if n_object.get('notification_format') == default_notification_format_for_watch:
n_object['notification_format'] = default_app_settings_notification_format
html_colour_enable = False
# HTML needs linebreak, but MarkDown and Text can use a linefeed
if n_object.get('notification_format') == 'HTML':
line_feed_sep = "<br>"
# Snapshot will be plaintext on the disk, convert to some kind of HTML
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
elif n_object.get('notification_format') == 'HTML Color':
line_feed_sep = "<br>"
# Snapshot will be plaintext on the disk, convert to some kind of HTML
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
html_colour_enable = True
else:
line_feed_sep = "\n"
triggered_text = ''
if len(trigger_text):
from . import html_tools
triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
if triggered_text:
triggered_text = line_feed_sep.join(triggered_text)
# Could be called as a 'test notification' with only 1 snapshot available
prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
current_snapshot = "Example text: example test\nExample text: change detection is fantastic\nExample text: even more examples\nExample text: a lot more examples"
if len(dates) > 1:
prev_snapshot = watch.get_history_snapshot(dates[-2])
current_snapshot = watch.get_history_snapshot(dates[-1])
n_object.update({
'current_snapshot': snapshot_contents,
'diff': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, html_colour=html_colour_enable),
'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=line_feed_sep),
'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': time.time(),
'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,
})
if watch:
n_object.update(watch.extra_notification_token_values())
return n_object
class update_worker(threading.Thread):
current_uuid = None
@@ -27,75 +98,8 @@ class update_worker(threading.Thread):
super().__init__(*args, **kwargs)
def queue_notification_for_watch(self, notification_q, n_object, watch):
from changedetectionio import diff
from changedetectionio.notification import default_notification_format_for_watch
dates = []
trigger_text = ''
now = time.time()
if watch:
watch_history = watch.history
dates = list(watch_history.keys())
trigger_text = watch.get('trigger_text', [])
# Add text that was triggered
if len(dates):
snapshot_contents = watch.get_history_snapshot(dates[-1])
else:
snapshot_contents = "No snapshot/history available, the watch should fetch atleast once."
# If we ended up here with "System default"
if n_object.get('notification_format') == default_notification_format_for_watch:
n_object['notification_format'] = self.datastore.data['settings']['application'].get('notification_format')
html_colour_enable = False
# HTML needs linebreak, but MarkDown and Text can use a linefeed
if n_object.get('notification_format') == 'HTML':
line_feed_sep = "<br>"
# Snapshot will be plaintext on the disk, convert to some kind of HTML
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
elif n_object.get('notification_format') == 'HTML Color':
line_feed_sep = "<br>"
# Snapshot will be plaintext on the disk, convert to some kind of HTML
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
html_colour_enable = True
else:
line_feed_sep = "\n"
triggered_text = ''
if len(trigger_text):
from . import html_tools
triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
if triggered_text:
triggered_text = line_feed_sep.join(triggered_text)
# Could be called as a 'test notification' with only 1 snapshot available
prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
current_snapshot = "Example text: example test\nExample text: change detection is fantastic\nExample text: even more examples\nExample text: a lot more examples"
if len(dates) > 1:
prev_snapshot = watch.get_history_snapshot(dates[-2])
current_snapshot = watch.get_history_snapshot(dates[-1])
n_object.update({
'current_snapshot': snapshot_contents,
'diff': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, html_colour=html_colour_enable),
'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=line_feed_sep),
'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,
})
if watch:
n_object.update(watch.extra_notification_token_values())
n_object = build_notification_object_for_watch(watch, n_object, self.datastore.data['settings']['application'].get('notification_format'))
logger.trace(f"Main rendered notification placeholders (diff_added etc) calculated in {time.time()-now:.3f}s")
logger.debug("Queued notification for sending")
notification_q.put(n_object)