From 5b5449e034686d4b39299257a2862bed2bf63445 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Sun, 26 Oct 2025 00:41:12 +0200 Subject: [PATCH] Tidy tests and word_diff handling --- changedetectionio/api/Watch.py | 14 +++- changedetectionio/diff/__init__.py | 72 +++++++++++----- changedetectionio/notification/handler.py | 15 +++- changedetectionio/tests/test_notification.py | 7 +- .../tests/unit/test_jinja2_security.py | 82 ------------------- 5 files changed, 81 insertions(+), 109 deletions(-) diff --git a/changedetectionio/api/Watch.py b/changedetectionio/api/Watch.py index 8ef684dcd..9446ccfa4 100644 --- a/changedetectionio/api/Watch.py +++ b/changedetectionio/api/Watch.py @@ -262,10 +262,20 @@ class WatchHistoryDiff(Resource): ) mimetype = "text/html" if output_format == 'html' else "text/plain" + import re if 'html' in output_format: - content = content.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '
\r\n') + content = re.sub( + re.escape(CUSTOM_LINEBREAK_PLACEHOLDER) + r'\r?\n?', + '
\\r\\n', + content + ) else: - content = content.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '\r\n') + # texty types + content = re.sub( + re.escape(CUSTOM_LINEBREAK_PLACEHOLDER) + r'\r?\n?', + '\\r\\n', + content + ) response = make_response(content, 200) response.mimetype = mimetype diff --git a/changedetectionio/diff/__init__.py b/changedetectionio/diff/__init__.py index cfd7d7f2c..6fbf74a87 100644 --- a/changedetectionio/diff/__init__.py +++ b/changedetectionio/diff/__init__.py @@ -110,27 +110,61 @@ def render_inline_word_diff(before_line: str, after_line: str, ignore_junk: bool whole_line_replaced = not any(op == 0 and text.strip() for op, text in diffs) # Build the output using placemarkers - result_parts = [] + # When whole line is replaced, wrap entire removed content once and entire added content once + if whole_line_replaced: + removed_tokens = [] + added_tokens = [] - for op, text in diffs: - if op == 0: # Equal - result_parts.append(text) - elif op == 1: # Insertion - content = text.rstrip() - trailing = text[len(content):] if len(text) > len(content) else '' - line_break = '\n' if whole_line_replaced else '' - placemarker_open = CHANGED_INTO_PLACEMARKER_OPEN if whole_line_replaced else ADDED_PLACEMARKER_OPEN - placemarker_closed = CHANGED_INTO_PLACEMARKER_CLOSED if whole_line_replaced else ADDED_PLACEMARKER_CLOSED - result_parts.append(f'{placemarker_open}{content}{placemarker_closed}{trailing}{line_break}') - elif op == -1: # Deletion - content = text.rstrip() - trailing = text[len(content):] if len(text) > len(content) else '' - line_break = '\n' if whole_line_replaced else '' - placemarker_open = CHANGED_PLACEMARKER_OPEN if whole_line_replaced else REMOVED_PLACEMARKER_OPEN - placemarker_closed = CHANGED_PLACEMARKER_CLOSED if whole_line_replaced else REMOVED_PLACEMARKER_CLOSED - result_parts.append(f'{placemarker_open}{content}{placemarker_closed}{trailing}{line_break}') + for op, text in diffs: + if op == 0: # Equal (e.g., whitespace tokens in common positions) + # Include in both removed and added to preserve spacing + removed_tokens.append(text) + added_tokens.append(text) + elif op == -1: # Deletion + removed_tokens.append(text) + elif op == 1: # Insertion + added_tokens.append(text) - return ''.join(result_parts), has_changes + # Join all tokens and wrap the entire string once for removed, once for added + result_parts = [] + + if removed_tokens: + removed_full = ''.join(removed_tokens).rstrip() + trailing_removed = ''.join(removed_tokens)[len(removed_full):] if len(''.join(removed_tokens)) > len(removed_full) else '' + result_parts.append(f'{CHANGED_PLACEMARKER_OPEN}{removed_full}{CHANGED_PLACEMARKER_CLOSED}{trailing_removed}') + + if added_tokens: + if result_parts: # Add newline between removed and added + result_parts.append('\n') + added_full = ''.join(added_tokens).rstrip() + trailing_added = ''.join(added_tokens)[len(added_full):] if len(''.join(added_tokens)) > len(added_full) else '' + result_parts.append(f'{CHANGED_INTO_PLACEMARKER_OPEN}{added_full}{CHANGED_INTO_PLACEMARKER_CLOSED}{trailing_added}') + + return ''.join(result_parts), has_changes + else: + # Inline changes within the line + result_parts = [] + for op, text in diffs: + if op == 0: # Equal + result_parts.append(text) + elif op == 1: # Insertion + # Don't wrap empty content (e.g., whitespace-only tokens after rstrip) + content = text.rstrip() + trailing = text[len(content):] if len(text) > len(content) else '' + if content: + result_parts.append(f'{ADDED_PLACEMARKER_OPEN}{content}{ADDED_PLACEMARKER_CLOSED}{trailing}') + else: + result_parts.append(trailing) + elif op == -1: # Deletion + # Don't wrap empty content (e.g., whitespace-only tokens after rstrip) + content = text.rstrip() + trailing = text[len(content):] if len(text) > len(content) else '' + if content: + result_parts.append(f'{REMOVED_PLACEMARKER_OPEN}{content}{REMOVED_PLACEMARKER_CLOSED}{trailing}') + else: + result_parts.append(trailing) + + return ''.join(result_parts), has_changes def render_nested_line_diff(before_line: str, after_line: str, ignore_junk: bool = False, tokenizer: str = 'words_and_html') -> tuple[str, str, bool]: diff --git a/changedetectionio/notification/handler.py b/changedetectionio/notification/handler.py index 1aa1498db..47c6b3bf0 100644 --- a/changedetectionio/notification/handler.py +++ b/changedetectionio/notification/handler.py @@ -1,5 +1,6 @@ import time +import re import apprise from apprise import NotifyFormat from loguru import logger @@ -365,11 +366,21 @@ def process_notification(n_object: NotificationContextData, datastore): apprise_input_format = NotifyFormat.HTML.value # Changed from MARKDOWN to HTML # Could have arrived at any stage, so we dont end up running .escape on it + # Replace CUSTOM_LINEBREAK_PLACEHOLDER followed by optional \r and/or \n if 'html' in requested_output_format: - n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '
\r\n') + # could be @BR@ with optional \r\n, so we dont add more \n's + n_body = re.sub( + re.escape(CUSTOM_LINEBREAK_PLACEHOLDER) + r'\r?\n?', + '
\\r\\n', + n_body + ) else: # texty types - n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '\r\n') + n_body = re.sub( + re.escape(CUSTOM_LINEBREAK_PLACEHOLDER) + r'\r?\n?', + '\\r\\n', + n_body + ) sent_objs.append({'title': n_title, 'body': n_body, diff --git a/changedetectionio/tests/test_notification.py b/changedetectionio/tests/test_notification.py index b5c0dfbf6..658d1bc78 100644 --- a/changedetectionio/tests/test_notification.py +++ b/changedetectionio/tests/test_notification.py @@ -16,7 +16,7 @@ from changedetectionio.notification import ( default_notification_title, valid_notification_formats, ) -from ..diff import HTML_CHANGED_STYLE, DIFF_HTML_LABEL_REMOVED +from ..diff import HTML_CHANGED_STYLE # Hard to just add more live server URLs when one test is already running (I think) @@ -547,12 +547,11 @@ def _test_color_notifications(client, notification_body_token): assert b'Queued 1 watch for rechecking.' in res.data wait_for_all_checks(client) - time.sleep(3) + time.sleep(2) with open("test-datastore/notification.txt", 'r') as f: x = f.read() - assert DIFF_HTML_LABEL_REMOVED.format(content='Which is across multiple lines') in x - s = f'Which is across multiple lines' + s = f'Which is across multiple lines
' assert s in x client.get( diff --git a/changedetectionio/tests/unit/test_jinja2_security.py b/changedetectionio/tests/unit/test_jinja2_security.py index bc5c9792c..6a56604db 100644 --- a/changedetectionio/tests/unit/test_jinja2_security.py +++ b/changedetectionio/tests/unit/test_jinja2_security.py @@ -55,88 +55,6 @@ class TestJinja2SSTI(unittest.TestCase): x = safe_jinja.render_fully_escaped('woo dfdfd') self.assertEqual(x, "woo <a href="https://google.com">dfdfd</a>") - def test_diff_unescape_difference_spans_filter_security(self): - """Test that diff_unescape_difference_spans filter only allows trusted diff spans and blocks XSS.""" - import re - from markupsafe import Markup, escape - - # Import the constants from diff module - from changedetectionio.diff import REMOVED_STYLE, ADDED_STYLE, DIFF_HTML_LABEL_REMOVED, DIFF_HTML_LABEL_ADDED, DIFF_HTML_LABEL_INSERTED - - # Recreate the filter logic for testing - def diff_unescape_difference_spans(content): - if not content: - return Markup('') - - # Step 1: Escape everything like Jinja2 would (XSS protection) - escaped_content = escape(str(content)) - - # Step 2: Selectively unescape only trusted diff spans - result = re.sub( - rf'<span style="({REMOVED_STYLE}|{ADDED_STYLE})" title="([A-Za-z0-9]+)">', - r'', - str(escaped_content), - flags=re.IGNORECASE - ) - - # Unescape closing tags (balanced) - open_count = result.count('', 1) - - return Markup(result) - - # Test 1: Valid diff spans should be unescaped - valid_diff_content = f'{DIFF_HTML_LABEL_REMOVED.format(content="old text")}\n{DIFF_HTML_LABEL_INSERTED.format(content="new text")}' - result = diff_unescape_difference_spans(valid_diff_content) - self.assertIn('alert("xss")', - 'evil', - 'text', - '', - '">' - ] - - for xss_attempt in xss_attempts: - result = diff_unescape_difference_spans(xss_attempt) - # Key security test: Should not contain EXECUTABLE HTML/JS - # (the critical thing is that < and > are escaped, preventing execution) - self.assertNotIn('' - result = diff_unescape_difference_spans(mixed_content) - self.assertIn('', str(result)) # No actual script tag if __name__ == '__main__':