Compare commits

...

13 Commits

14 changed files with 96 additions and 86 deletions

View File

@@ -2,7 +2,7 @@
# Test that we can still build on Alpine (musl modified libc https://musl.libc.org/)
# Some packages wont install via pypi because they dont have a wheel available under this architecture.
FROM ghcr.io/linuxserver/baseimage-alpine:3.18
FROM ghcr.io/linuxserver/baseimage-alpine:3.21
ENV PYTHONUNBUFFERED=1
COPY requirements.txt /requirements.txt
@@ -28,6 +28,6 @@ RUN \
py3-pip && \
echo "**** pip3 install test of changedetection.io ****" && \
pip3 install -U pip wheel setuptools && \
pip3 install -U --no-cache-dir --find-links https://wheel-index.linuxserver.io/alpine-3.18/ -r /requirements.txt && \
pip3 install -U --no-cache-dir --find-links https://wheel-index.linuxserver.io/alpine-3.21/ -r /requirements.txt && \
apk del --purge \
build-dependencies

View File

@@ -2,7 +2,7 @@
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
__version__ = '0.48.01'
__version__ = '0.48.05'
from changedetectionio.strtobool import strtobool
from json.decoder import JSONDecodeError

View File

@@ -1,6 +1,9 @@
import difflib
from typing import List, Iterator, Union
REMOVED_STYLE = "background-color: #fadad7; color: #b30000;"
ADDED_STYLE = "background-color: #eaf2c2; color: #406619;"
def same_slicer(lst: List[str], start: int, end: int) -> List[str]:
"""Return a slice of the list, or a single element if start == end."""
return lst[start:end] if start != end else [lst[start]]
@@ -33,24 +36,26 @@ def customSequenceMatcher(
"""
cruncher = difflib.SequenceMatcher(isjunk=lambda x: x in " \t", a=before, b=after)
for tag, alo, ahi, blo, bhi in cruncher.get_opcodes():
if include_equal and tag == 'equal':
yield before[alo:ahi]
elif include_removed and tag == 'delete':
if html_colour:
yield [f'<span style="background-color: #ffcecb;">{line}</span>' for line in same_slicer(before, alo, ahi)]
yield [f'<span 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="background-color: #ffcecb;">{line}</span>' for line in same_slicer(before, alo, ahi)] + \
[f'<span style="background-color: #dafbe1;">{line}</span>' for line in same_slicer(after, blo, bhi)]
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)]
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="background-color: #dafbe1;">{line}</span>' for line in same_slicer(after, blo, bhi)]
yield [f'<span 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)

View File

@@ -247,37 +247,32 @@ class model(watch_base):
bump = self.history
return self.__newest_history_key
# Given an arbitrary timestamp, find the closest next key
# For example, last_viewed = 1000 so it should return the next 1001 timestamp
#
# used for the [diff] button so it can preset a smarter from_version
# Given an arbitrary timestamp, find the best history key for the [diff] button so it can preset a smarter from_version
@property
def get_next_snapshot_key_to_last_viewed(self):
def get_from_version_based_on_last_viewed(self):
"""Unfortunately for now timestamp is stored as string key"""
keys = list(self.history.keys())
if not keys:
return None
if len(keys) == 1:
return keys[0]
last_viewed = int(self.get('last_viewed'))
prev_k = keys[0]
sorted_keys = sorted(keys, key=lambda x: int(x))
sorted_keys.reverse()
# When the 'last viewed' timestamp is greater than the newest snapshot, return second last
if last_viewed > int(sorted_keys[0]):
# When the 'last viewed' timestamp is greater than or equal the newest snapshot, return second newest
if last_viewed >= int(sorted_keys[0]):
return sorted_keys[1]
# When the 'last viewed' timestamp is between snapshots, return the older snapshot
for newer, older in list(zip(sorted_keys[0:], sorted_keys[1:])):
if last_viewed < int(newer) and last_viewed >= int(older):
return older
for k in sorted_keys:
if int(k) < last_viewed:
if prev_k == sorted_keys[0]:
# Return the second last one so we dont recommend the same version compares itself
return sorted_keys[1]
return prev_k
prev_k = k
return keys[0]
# When the 'last viewed' timestamp is less than the oldest snapshot, return oldest
return sorted_keys[-1]
def get_history_snapshot(self, timestamp):
import brotli

View File

@@ -23,7 +23,7 @@ valid_tokens = {
}
default_notification_format_for_watch = 'System default'
default_notification_format = 'Text'
default_notification_format = 'HTML Color'
default_notification_body = '{{watch_url}} had a change.\n---\n{{diff}}\n---\n'
default_notification_title = 'ChangeDetection.io Notification - {{watch_url}}'

View File

@@ -33,8 +33,8 @@ class difference_detection_processor():
url = self.watch.link
# Protect against file://, file:/ access, check the real "link" without any meta "source:" etc prepended.
if re.search(r'^file:/', url.strip(), re.IGNORECASE):
# Protect against file:, file:/, file:// access, check the real "link" without any meta "source:" etc prepended.
if re.search(r'^file:', url.strip(), re.IGNORECASE):
if not strtobool(os.getenv('ALLOW_FILE_URI', 'false')):
raise Exception(
"file:// type access is denied for security reasons."

View File

@@ -191,7 +191,7 @@
{% if watch.history_n >= 2 %}
{% if is_unviewed %}
<a href="{{ url_for('diff_history_page', uuid=watch.uuid, from_version=watch.get_next_snapshot_key_to_last_viewed) }}" target="{{watch.uuid}}" class="pure-button pure-button-primary diff-link">History</a>
<a href="{{ url_for('diff_history_page', uuid=watch.uuid, from_version=watch.get_from_version_based_on_last_viewed) }}" target="{{watch.uuid}}" class="pure-button pure-button-primary diff-link">History</a>
{% else %}
<a href="{{ url_for('diff_history_page', uuid=watch.uuid)}}" target="{{watch.uuid}}" class="pure-button pure-button-primary diff-link">History</a>
{% endif %}

View File

@@ -113,7 +113,8 @@ def test_check_add_line_contains_trigger(client, live_server, measure_memory_usa
res = client.post(
url_for("settings_page"),
data={"application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }}",
"application-notification_body": 'triggered text was -{{triggered_text}}- 网站监测 内容更新了',
# triggered_text will contain multiple lines
"application-notification_body": 'triggered text was -{{triggered_text}}- ### 网站监测 内容更新了 ####',
# https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
"application-notification_urls": test_notification_url,
"application-minutes_between_check": 180,
@@ -171,7 +172,7 @@ def test_check_add_line_contains_trigger(client, live_server, measure_memory_usa
assert os.path.isfile("test-datastore/notification.txt"), "Notification fired because I can see the output file"
with open("test-datastore/notification.txt", 'rb') as f:
response = f.read()
assert b'-Oh yes please-' in response
assert b'-Oh yes please' in response
assert '网站监测 内容更新了'.encode('utf-8') in response
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)

View File

@@ -442,9 +442,9 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
assert b"Error: You must have atleast one watch configured for 'test notification' to work" in res.data
def test_html_color_notifications(client, live_server, measure_memory_usage):
def _test_color_notifications(client, notification_body_token):
#live_server_setup(live_server)
from changedetectionio.diff import ADDED_STYLE, REMOVED_STYLE
set_original_response()
@@ -461,7 +461,7 @@ def test_html_color_notifications(client, live_server, measure_memory_usage):
data={
"application-fetch_backend": "html_requests",
"application-minutes_between_check": 180,
"application-notification_body": '{{diff}}',
"application-notification_body": notification_body_token,
"application-notification_format": "HTML Color",
"application-notification_urls": test_notification_url,
"application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }}",
@@ -492,7 +492,7 @@ def test_html_color_notifications(client, live_server, measure_memory_usage):
with open("test-datastore/notification.txt", 'r') as f:
x = f.read()
assert '<span style="background-color: #ffcecb;">Which is across multiple lines' in x
assert f'<span style="{REMOVED_STYLE}">Which is across multiple lines' in x
client.get(
@@ -500,3 +500,9 @@ def test_html_color_notifications(client, live_server, measure_memory_usage):
follow_redirects=True
)
def test_html_color_notifications(client, live_server, measure_memory_usage):
#live_server_setup(live_server)
_test_color_notifications(client, '{{diff}}')
_test_color_notifications(client, '{{diff_full}}')

View File

@@ -1,9 +1,7 @@
import os
from flask import url_for
from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks
import time
from .util import live_server_setup, wait_for_all_checks
from .. import strtobool
@@ -61,54 +59,44 @@ def test_bad_access(client, live_server, measure_memory_usage):
assert b'Watch protocol is not permitted by SAFE_PROTOCOL_REGEX' in res.data
def test_file_slashslash_access(client, live_server, measure_memory_usage):
#live_server_setup(live_server)
def _runner_test_various_file_slash(client, file_uri):
test_file_path = os.path.abspath(__file__)
# file:// is permitted by default, but it will be caught by ALLOW_FILE_URI
client.post(
url_for("form_quick_watch_add"),
data={"url": f"file://{test_file_path}", "tags": ''},
data={"url": file_uri, "tags": ''},
follow_redirects=True
)
wait_for_all_checks(client)
res = client.get(url_for("index"))
substrings = [b"URLs with hostname components are not permitted", b"No connection adapters were found for"]
# If it is enabled at test time
if strtobool(os.getenv('ALLOW_FILE_URI', 'false')):
res = client.get(
url_for("preview_page", uuid="first"),
follow_redirects=True
)
if file_uri.startswith('file:///'):
# This one should be the full qualified path to the file and should get the contents of this file
res = client.get(
url_for("preview_page", uuid="first"),
follow_redirects=True
)
assert b'_runner_test_various_file_slash' in res.data
else:
# This will give some error from requests or if it went to chrome, will give some other error :-)
assert any(s in res.data for s in substrings)
assert b"test_file_slashslash_access" in res.data
else:
# Default should be here
assert b'file:// type access is denied for security reasons.' in res.data
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data
def test_file_slash_access(client, live_server, measure_memory_usage):
#live_server_setup(live_server)
# file: is NOT permitted by default, so it will be caught by ALLOW_FILE_URI check
test_file_path = os.path.abspath(__file__)
# file:// is permitted by default, but it will be caught by ALLOW_FILE_URI
client.post(
url_for("form_quick_watch_add"),
data={"url": f"file:/{test_file_path}", "tags": ''},
follow_redirects=True
)
wait_for_all_checks(client)
res = client.get(url_for("index"))
# If it is enabled at test time
if strtobool(os.getenv('ALLOW_FILE_URI', 'false')):
# So it should permit it, but it should fall back to the 'requests' library giving an error
# (but means it gets passed to playwright etc)
assert b"URLs with hostname components are not permitted" in res.data
else:
# Default should be here
assert b'file:// type access is denied for security reasons.' in res.data
_runner_test_various_file_slash(client, file_uri=f"file://{test_file_path}")
_runner_test_various_file_slash(client, file_uri=f"file:/{test_file_path}")
_runner_test_various_file_slash(client, file_uri=f"file:{test_file_path}") # CVE-2024-56509
def test_xss(client, live_server, measure_memory_usage):
#live_server_setup(live_server)

View File

@@ -16,7 +16,6 @@ class TestDiffBuilder(unittest.TestCase):
watch = Watch.model(datastore_path='/tmp', default={})
watch.ensure_data_dir_exists()
watch['last_viewed'] = 110
# Contents from the browser are always returned from the browser/requests/etc as str, str is basically UTF-16 in python
watch.save_history_text(contents="hello world", timestamp=100, snapshot_id=str(uuid_builder.uuid4()))
@@ -25,31 +24,42 @@ class TestDiffBuilder(unittest.TestCase):
watch.save_history_text(contents="hello world", timestamp=112, snapshot_id=str(uuid_builder.uuid4()))
watch.save_history_text(contents="hello world", timestamp=115, snapshot_id=str(uuid_builder.uuid4()))
watch.save_history_text(contents="hello world", timestamp=117, snapshot_id=str(uuid_builder.uuid4()))
p = watch.get_from_version_based_on_last_viewed
assert p == "100", "Correct 'last viewed' timestamp was detected"
p = watch.get_next_snapshot_key_to_last_viewed
assert p == "112", "Correct last-viewed timestamp was detected"
watch['last_viewed'] = 110
p = watch.get_from_version_based_on_last_viewed
assert p == "109", "Correct 'last viewed' timestamp was detected"
# When there is only one step of difference from the end of the list, it should return second-last change
watch['last_viewed'] = 116
p = watch.get_next_snapshot_key_to_last_viewed
assert p == "115", "Correct 'second last' last-viewed timestamp was detected when using the last timestamp"
p = watch.get_from_version_based_on_last_viewed
assert p == "115", "Correct 'last viewed' timestamp was detected"
watch['last_viewed'] = 99
p = watch.get_next_snapshot_key_to_last_viewed
assert p == "100"
p = watch.get_from_version_based_on_last_viewed
assert p == "100", "When the 'last viewed' timestamp is less than the oldest snapshot, return oldest"
watch['last_viewed'] = 200
p = watch.get_next_snapshot_key_to_last_viewed
assert p == "115", "When the 'last viewed' timestamp is greater than the newest snapshot, return second last "
p = watch.get_from_version_based_on_last_viewed
assert p == "115", "When the 'last viewed' timestamp is greater than the newest snapshot, return second newest"
watch['last_viewed'] = 109
p = watch.get_next_snapshot_key_to_last_viewed
p = watch.get_from_version_based_on_last_viewed
assert p == "109", "Correct when its the same time"
# new empty one
watch = Watch.model(datastore_path='/tmp', default={})
p = watch.get_next_snapshot_key_to_last_viewed
p = watch.get_from_version_based_on_last_viewed
assert p == None, "None when no history available"
watch.save_history_text(contents="hello world", timestamp=100, snapshot_id=str(uuid_builder.uuid4()))
p = watch.get_from_version_based_on_last_viewed
assert p == "100", "Correct with only one history snapshot"
watch['last_viewed'] = 200
p = watch.get_from_version_based_on_last_viewed
assert p == "100", "Correct with only one history snapshot"
if __name__ == '__main__':
unittest.main()

View File

@@ -28,6 +28,8 @@ class update_worker(threading.Thread):
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 = ''
@@ -44,6 +46,10 @@ class update_worker(threading.Thread):
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':
@@ -77,7 +83,7 @@ class update_worker(threading.Thread):
'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),
'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,

View File

@@ -12,9 +12,6 @@ services:
# environment:
# Default listening port, can also be changed with the -p option
# - PORT=5000
# - PUID=1000
# - PGID=1000
#
# Log levels are in descending order. (TRACE is the most detailed one)
# Log output levels: TRACE, DEBUG(default), INFO, SUCCESS, WARNING, ERROR, CRITICAL

View File

@@ -95,3 +95,5 @@ babel
# Needed for > 3.10, https://github.com/microsoft/playwright-python/issues/2096
greenlet >= 3.0.3
# Scheduler - Windows seemed to miss a lot of default timezone info (even "UTC" !)
tzdata