Self-Hosted
/
changedetection.io
mirror of https://github.com/dgtlmoon/changedetection.io.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

New setting to allow passwordless access to your 'diff' page - perfect for sharing your diff page securely, refactored login code (#1357)

Browse Source
pull/1367/head
dgtlmoon 1 year ago committed by GitHub
parent f8e587c415
commit 0ed897c50f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 199 additions and 75 deletions
Show Stats Download Patch File Download Diff File

4
.github/workflows/test-only.yml vendored
Unescape View File

@ -67,10 +67,10 @@ jobs:
sleep 3 sleep 3
# Should return 0 (no error) when grep finds it # Should return 0 (no error) when grep finds it
curl -s http://localhost:5556 |grep -q checkbox-uuid curl -s http://localhost:5556 |grep -q checkbox-uuid
curl -s http://localhost:5556/rss|grep -q rss-specification
# and IPv6 # and IPv6
curl -s -g -6 "http://[::1]:5556"|grep -q checkbox-uuid curl -s -g -6 "http://[::1]:5556"|grep -q checkbox-uuid
curl -s -g -6 "http://[::1]:5556/rss"|grep -q rss-specification
#export WEBDRIVER_URL=http://localhost:4444/wd/hub #export WEBDRIVER_URL=http://localhost:4444/wd/hub
#pytest tests/fetchers/test_content.py #pytest tests/fetchers/test_content.py

128
changedetectionio/__init__.py
Unescape View File

@ -1,5 +1,15 @@
#!/usr/bin/python3 #!/usr/bin/python3
from changedetectionio import queuedWatchMetaData
from copy import deepcopy
from distutils.util import strtobool
from feedgen.feed import FeedGenerator
from flask_compress import Compress as FlaskCompress
from flask_login import current_user
from flask_restful import abort, Api
from flask_wtf import CSRFProtect
from functools import wraps
from threading import Event
import datetime import datetime
import flask_login import flask_login
import logging import logging
@ -10,12 +20,6 @@ import threading
import time import time
import timeago import timeago
from changedetectionio import queuedWatchMetaData
from copy import deepcopy
from distutils.util import strtobool
from feedgen.feed import FeedGenerator
from threading import Event
from flask import ( from flask import (
Flask, Flask,
abort, abort,
@ -28,10 +32,6 @@ from flask import (
session, session,
url_for, url_for,
) )
from flask_compress import Compress as FlaskCompress
from flask_login import login_required
from flask_restful import abort, Api
from flask_wtf import CSRFProtect
from changedetectionio import html_tools from changedetectionio import html_tools
from changedetectionio.api import api_v1 from changedetectionio.api import api_v1
@ -65,8 +65,6 @@ app.config.exit = Event()
app.config['NEW_VERSION_AVAILABLE'] = False app.config['NEW_VERSION_AVAILABLE'] = False
app.config['LOGIN_DISABLED'] = False
#app.config["EXPLAIN_TEMPLATE_LOADING"] = True #app.config["EXPLAIN_TEMPLATE_LOADING"] = True
# Disables caching of the templates # Disables caching of the templates
@ -148,7 +146,6 @@ class User(flask_login.UserMixin):
# Compare given password against JSON store or Env var # Compare given password against JSON store or Env var
def check_password(self, password): def check_password(self, password):
import base64 import base64
import hashlib import hashlib
@ -156,11 +153,9 @@ class User(flask_login.UserMixin):
raw_salt_pass = os.getenv("SALTED_PASS", False) raw_salt_pass = os.getenv("SALTED_PASS", False)
if not raw_salt_pass: if not raw_salt_pass:
raw_salt_pass = datastore.data['settings']['application']['password'] raw_salt_pass = datastore.data['settings']['application'].get('password')
raw_salt_pass = base64.b64decode(raw_salt_pass) raw_salt_pass = base64.b64decode(raw_salt_pass)
salt_from_storage = raw_salt_pass[:32] # 32 is the length of the salt salt_from_storage = raw_salt_pass[:32] # 32 is the length of the salt
# Use the exact same setup you used to generate the key, but this time put in the password to check # Use the exact same setup you used to generate the key, but this time put in the password to check
@ -170,21 +165,44 @@ class User(flask_login.UserMixin):
salt_from_storage, salt_from_storage,
100000 100000
) )
new_key = salt_from_storage + new_key new_key = salt_from_storage + new_key
return new_key == raw_salt_pass return new_key == raw_salt_pass
pass pass
def login_optionally_required(func):
@wraps(func)
def decorated_view(*args, **kwargs):
has_password_enabled = datastore.data['settings']['application'].get('password') or os.getenv("SALTED_PASS", False)
# Permitted
if request.endpoint == 'static_content' and request.view_args['group'] == 'styles':
return func(*args, **kwargs)
# Permitted
elif request.endpoint == 'diff_history_page' and datastore.data['settings']['application'].get('shared_diff_access'):
return func(*args, **kwargs)
elif request.method in flask_login.config.EXEMPT_METHODS:
return func(*args, **kwargs)
elif app.config.get('LOGIN_DISABLED'):
return func(*args, **kwargs)
elif has_password_enabled and not current_user.is_authenticated:
return app.login_manager.unauthorized()
return func(*args, **kwargs)
return decorated_view
def changedetection_app(config=None, datastore_o=None): def changedetection_app(config=None, datastore_o=None):
global datastore global datastore
datastore = datastore_o datastore = datastore_o
# so far just for read-only via tests, but this will be moved eventually to be the main source # so far just for read-only via tests, but this will be moved eventually to be the main source
# (instead of the global var) # (instead of the global var)
app.config['DATASTORE']=datastore_o app.config['DATASTORE'] = datastore_o
#app.config.update(config or {})
login_manager = flask_login.LoginManager(app) login_manager = flask_login.LoginManager(app)
login_manager.login_view = 'login' login_manager.login_view = 'login'
@ -212,6 +230,8 @@ def changedetection_app(config=None, datastore_o=None):
# https://flask-cors.readthedocs.io/en/latest/ # https://flask-cors.readthedocs.io/en/latest/
# CORS(app) # CORS(app)
@login_manager.user_loader @login_manager.user_loader
def user_loader(email): def user_loader(email):
user = User() user = User()
@ -220,7 +240,7 @@ def changedetection_app(config=None, datastore_o=None):
@login_manager.unauthorized_handler @login_manager.unauthorized_handler
def unauthorized_handler(): def unauthorized_handler():
# @todo validate its a URL of this host and use that flash("You must be logged in, please log in.", 'error')
return redirect(url_for('login', next=url_for('index'))) return redirect(url_for('login', next=url_for('index')))
@app.route('/logout') @app.route('/logout')
@ -233,10 +253,6 @@ def changedetection_app(config=None, datastore_o=None):
@app.route('/login', methods=['GET', 'POST']) @app.route('/login', methods=['GET', 'POST'])
def login(): def login():
if not datastore.data['settings']['application']['password'] and not os.getenv("SALTED_PASS", False):
flash("Login not required, no password enabled.", "notice")
return redirect(url_for('index'))
if request.method == 'GET': if request.method == 'GET':
if flask_login.current_user.is_authenticated: if flask_login.current_user.is_authenticated:
flash("Already logged in") flash("Already logged in")
@ -271,27 +287,22 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('login')) return redirect(url_for('login'))
@app.before_request @app.before_request
def do_something_whenever_a_request_comes_in(): def before_request_handle_cookie_x_settings():
# Disable password login if there is not one set
# (No password in settings or env var)
app.config['LOGIN_DISABLED'] = datastore.data['settings']['application']['password'] == False and os.getenv("SALTED_PASS", False) == False
# Set the auth cookie path if we're running as X-settings/X-Forwarded-Prefix # Set the auth cookie path if we're running as X-settings/X-Forwarded-Prefix
if os.getenv('USE_X_SETTINGS') and 'X-Forwarded-Prefix' in request.headers: if os.getenv('USE_X_SETTINGS') and 'X-Forwarded-Prefix' in request.headers:
app.config['REMEMBER_COOKIE_PATH'] = request.headers['X-Forwarded-Prefix'] app.config['REMEMBER_COOKIE_PATH'] = request.headers['X-Forwarded-Prefix']
app.config['SESSION_COOKIE_PATH'] = request.headers['X-Forwarded-Prefix'] app.config['SESSION_COOKIE_PATH'] = request.headers['X-Forwarded-Prefix']
# For the RSS path, allow access via a token return None
if request.path == '/rss' and request.args.get('token'):
app_rss_token = datastore.data['settings']['application']['rss_access_token']
rss_url_token = request.args.get('token')
if app_rss_token == rss_url_token:
app.config['LOGIN_DISABLED'] = True
@app.route("/rss", methods=['GET']) @app.route("/rss", methods=['GET'])
@login_required
def rss(): def rss():
# Always requires token set
app_rss_token = datastore.data['settings']['application'].get('rss_access_token')
rss_url_token = request.args.get('token')
if rss_url_token != app_rss_token:
return "Access denied, bad token", 403
from . import diff from . import diff
limit_tag = request.args.get('tag') limit_tag = request.args.get('tag')
@ -365,7 +376,7 @@ def changedetection_app(config=None, datastore_o=None):
return response return response
@app.route("/", methods=['GET']) @app.route("/", methods=['GET'])
@login_required @login_optionally_required
def index(): def index():
from changedetectionio import forms from changedetectionio import forms
@ -429,7 +440,7 @@ def changedetection_app(config=None, datastore_o=None):
# AJAX endpoint for sending a test # AJAX endpoint for sending a test
@app.route("/notification/send-test", methods=['POST']) @app.route("/notification/send-test", methods=['POST'])
@login_required @login_optionally_required
def ajax_callback_send_notification_test(): def ajax_callback_send_notification_test():
import apprise import apprise
@ -462,7 +473,7 @@ def changedetection_app(config=None, datastore_o=None):
@app.route("/clear_history/<string:uuid>", methods=['GET']) @app.route("/clear_history/<string:uuid>", methods=['GET'])
@login_required @login_optionally_required
def clear_watch_history(uuid): def clear_watch_history(uuid):
try: try:
datastore.clear_watch_history(uuid) datastore.clear_watch_history(uuid)
@ -474,7 +485,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index')) return redirect(url_for('index'))
@app.route("/clear_history", methods=['GET', 'POST']) @app.route("/clear_history", methods=['GET', 'POST'])
@login_required @login_optionally_required
def clear_all_history(): def clear_all_history():
if request.method == 'POST': if request.method == 'POST':
@ -531,7 +542,7 @@ def changedetection_app(config=None, datastore_o=None):
@app.route("/edit/<string:uuid>", methods=['GET', 'POST']) @app.route("/edit/<string:uuid>", methods=['GET', 'POST'])
@login_required @login_optionally_required
# https://stackoverflow.com/questions/42984453/wtforms-populate-form-with-data-if-data-exists # https://stackoverflow.com/questions/42984453/wtforms-populate-form-with-data-if-data-exists
# https://wtforms.readthedocs.io/en/3.0.x/forms/#wtforms.form.Form.populate_obj ? # https://wtforms.readthedocs.io/en/3.0.x/forms/#wtforms.form.Form.populate_obj ?
@ -677,7 +688,7 @@ def changedetection_app(config=None, datastore_o=None):
return output return output
@app.route("/settings", methods=['GET', "POST"]) @app.route("/settings", methods=['GET', "POST"])
@login_required @login_optionally_required
def settings_page(): def settings_page():
from changedetectionio import content_fetcher, forms from changedetectionio import content_fetcher, forms
@ -757,7 +768,7 @@ def changedetection_app(config=None, datastore_o=None):
return output return output
@app.route("/import", methods=['GET', "POST"]) @app.route("/import", methods=['GET', "POST"])
@login_required @login_optionally_required
def import_page(): def import_page():
remaining_urls = [] remaining_urls = []
if request.method == 'POST': if request.method == 'POST':
@ -795,7 +806,7 @@ def changedetection_app(config=None, datastore_o=None):
# Clear all statuses, so we do not see the 'unviewed' class # Clear all statuses, so we do not see the 'unviewed' class
@app.route("/form/mark-all-viewed", methods=['GET']) @app.route("/form/mark-all-viewed", methods=['GET'])
@login_required @login_optionally_required
def mark_all_viewed(): def mark_all_viewed():
# Save the current newest history as the most recently viewed # Save the current newest history as the most recently viewed
@ -805,7 +816,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index')) return redirect(url_for('index'))
@app.route("/diff/<string:uuid>", methods=['GET', 'POST']) @app.route("/diff/<string:uuid>", methods=['GET', 'POST'])
@login_required @login_optionally_required
def diff_history_page(uuid): def diff_history_page(uuid):
from changedetectionio import forms from changedetectionio import forms
@ -884,6 +895,10 @@ def changedetection_app(config=None, datastore_o=None):
is_html_webdriver = True if watch.get('fetch_backend') == 'html_webdriver' or ( is_html_webdriver = True if watch.get('fetch_backend') == 'html_webdriver' or (
watch.get('fetch_backend', None) is None and system_uses_webdriver) else False watch.get('fetch_backend', None) is None and system_uses_webdriver) else False
password_enabled_and_share_is_off = False
if datastore.data['settings']['application'].get('password') or os.getenv("SALTED_PASS", False):
password_enabled_and_share_is_off = not datastore.data['settings']['application'].get('shared_diff_access')
output = render_template("diff.html", output = render_template("diff.html",
current_diff_url=watch['url'], current_diff_url=watch['url'],
current_previous_version=str(previous_version), current_previous_version=str(previous_version),
@ -897,6 +912,7 @@ def changedetection_app(config=None, datastore_o=None):
left_sticky=True, left_sticky=True,
newest=newest_version_file_contents, newest=newest_version_file_contents,
newest_version_timestamp=dates[-1], newest_version_timestamp=dates[-1],
password_enabled_and_share_is_off=password_enabled_and_share_is_off,
previous=previous_version_file_contents, previous=previous_version_file_contents,
screenshot=screenshot_url, screenshot=screenshot_url,
uuid=uuid, uuid=uuid,
@ -907,7 +923,7 @@ def changedetection_app(config=None, datastore_o=None):
return output return output
@app.route("/preview/<string:uuid>", methods=['GET']) @app.route("/preview/<string:uuid>", methods=['GET'])
@login_required @login_optionally_required
def preview_page(uuid): def preview_page(uuid):
content = [] content = []
ignored_line_numbers = [] ignored_line_numbers = []
@ -997,7 +1013,7 @@ def changedetection_app(config=None, datastore_o=None):
return output return output
@app.route("/settings/notification-logs", methods=['GET']) @app.route("/settings/notification-logs", methods=['GET'])
@login_required @login_optionally_required
def notification_logs(): def notification_logs():
global notification_debug_log global notification_debug_log
output = render_template("notification-log.html", output = render_template("notification-log.html",
@ -1007,7 +1023,7 @@ def changedetection_app(config=None, datastore_o=None):
# We're good but backups are even better! # We're good but backups are even better!
@app.route("/backup", methods=['GET']) @app.route("/backup", methods=['GET'])
@login_required @login_optionally_required
def get_backup(): def get_backup():
import zipfile import zipfile
@ -1127,7 +1143,7 @@ def changedetection_app(config=None, datastore_o=None):
abort(404) abort(404)
@app.route("/form/add/quickwatch", methods=['POST']) @app.route("/form/add/quickwatch", methods=['POST'])
@login_required @login_optionally_required
def form_quick_watch_add(): def form_quick_watch_add():
from changedetectionio import forms from changedetectionio import forms
form = forms.quickWatchForm(request.form) form = forms.quickWatchForm(request.form)
@ -1159,7 +1175,7 @@ def changedetection_app(config=None, datastore_o=None):
@app.route("/api/delete", methods=['GET']) @app.route("/api/delete", methods=['GET'])
@login_required @login_optionally_required
def form_delete(): def form_delete():
uuid = request.args.get('uuid') uuid = request.args.get('uuid')
@ -1176,7 +1192,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index')) return redirect(url_for('index'))
@app.route("/api/clone", methods=['GET']) @app.route("/api/clone", methods=['GET'])
@login_required @login_optionally_required
def form_clone(): def form_clone():
uuid = request.args.get('uuid') uuid = request.args.get('uuid')
# More for testing, possible to return the first/only # More for testing, possible to return the first/only
@ -1191,7 +1207,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index')) return redirect(url_for('index'))
@app.route("/api/checknow", methods=['GET']) @app.route("/api/checknow", methods=['GET'])
@login_required @login_optionally_required
def form_watch_checknow(): def form_watch_checknow():
# Forced recheck will skip the 'skip if content is the same' rule (, 'reprocess_existing_data': True}))) # Forced recheck will skip the 'skip if content is the same' rule (, 'reprocess_existing_data': True})))
tag = request.args.get('tag') tag = request.args.get('tag')
@ -1225,7 +1241,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index', tag=tag)) return redirect(url_for('index', tag=tag))
@app.route("/form/checkbox-operations", methods=['POST']) @app.route("/form/checkbox-operations", methods=['POST'])
@login_required @login_optionally_required
def form_watch_list_checkbox_operations(): def form_watch_list_checkbox_operations():
op = request.form['op'] op = request.form['op']
uuids = request.form.getlist('uuids') uuids = request.form.getlist('uuids')
@ -1289,7 +1305,7 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index')) return redirect(url_for('index'))
@app.route("/api/share-url", methods=['GET']) @app.route("/api/share-url", methods=['GET'])
@login_required @login_optionally_required
def form_share_put_watch(): def form_share_put_watch():
"""Given a watch UUID, upload the info and return a share-link """Given a watch UUID, upload the info and return a share-link
the share-link can be imported/added""" the share-link can be imported/added"""

5
changedetectionio/blueprint/browser_steps/__init__.py
Unescape View File

@ -23,11 +23,10 @@
from distutils.util import strtobool from distutils.util import strtobool
from flask import Blueprint, request, make_response from flask import Blueprint, request, make_response
from flask_login import login_required
import os import os
import logging import logging
from changedetectionio.store import ChangeDetectionStore from changedetectionio.store import ChangeDetectionStore
from changedetectionio import login_optionally_required
browsersteps_live_ui_o = {} browsersteps_live_ui_o = {}
browsersteps_playwright_browser_interface = None browsersteps_playwright_browser_interface = None
browsersteps_playwright_browser_interface_browser = None browsersteps_playwright_browser_interface_browser = None
@ -65,7 +64,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
browser_steps_blueprint = Blueprint('browser_steps', __name__, template_folder="templates") browser_steps_blueprint = Blueprint('browser_steps', __name__, template_folder="templates")
@login_required @login_optionally_required
@browser_steps_blueprint.route("/browsersteps_update", methods=['GET', 'POST']) @browser_steps_blueprint.route("/browsersteps_update", methods=['GET', 'POST'])
def browsersteps_ui_update(): def browsersteps_ui_update():
import base64 import base64

12
changedetectionio/forms.py
Unescape View File

@ -459,17 +459,17 @@ class globalSettingsRequestForm(Form):
# datastore.data['settings']['application'].. # datastore.data['settings']['application']..
class globalSettingsApplicationForm(commonSettingsForm): class globalSettingsApplicationForm(commonSettingsForm):
api_access_token_enabled = BooleanField('API access token security check enabled', default=True, validators=[validators.Optional()])
base_url = StringField('Base URL', validators=[validators.Optional()]) base_url = StringField('Base URL', validators=[validators.Optional()])
global_subtractive_selectors = StringListField('Remove elements', [ValidateCSSJSONXPATHInput(allow_xpath=False, allow_json=False)]) empty_pages_are_a_change = BooleanField('Treat empty pages as a change?', default=False)
fetch_backend = RadioField('Fetch Method', default="html_requests", choices=content_fetcher.available_fetchers(), validators=[ValidateContentFetcherIsReady()])
global_ignore_text = StringListField('Ignore Text', [ValidateListRegex()]) global_ignore_text = StringListField('Ignore Text', [ValidateListRegex()])
global_subtractive_selectors = StringListField('Remove elements', [ValidateCSSJSONXPATHInput(allow_xpath=False, allow_json=False)])
ignore_whitespace = BooleanField('Ignore whitespace') ignore_whitespace = BooleanField('Ignore whitespace')
password = SaltyPasswordField()
removepassword_button = SubmitField('Remove password', render_kw={"class": "pure-button pure-button-primary"}) removepassword_button = SubmitField('Remove password', render_kw={"class": "pure-button pure-button-primary"})
empty_pages_are_a_change = BooleanField('Treat empty pages as a change?', default=False)
render_anchor_tag_content = BooleanField('Render anchor tag content', default=False) render_anchor_tag_content = BooleanField('Render anchor tag content', default=False)
fetch_backend = RadioField('Fetch Method', default="html_requests", choices=content_fetcher.available_fetchers(), validators=[ValidateContentFetcherIsReady()]) shared_diff_access = BooleanField('Allow access to view diff page when password is enabled', default=False, validators=[validators.Optional()])
api_access_token_enabled = BooleanField('API access token security check enabled', default=True, validators=[validators.Optional()])
password = SaltyPasswordField()
filter_failure_notification_threshold_attempts = IntegerField('Number of times the filter can be missing before sending a notification', filter_failure_notification_threshold_attempts = IntegerField('Number of times the filter can be missing before sending a notification',
render_kw={"style": "width: 5em;"}, render_kw={"style": "width: 5em;"},
validators=[validators.NumberRange(min=0, validators=[validators.NumberRange(min=0,

1
changedetectionio/model/App.py
Unescape View File

@ -40,6 +40,7 @@ class model(dict):
'notification_body': default_notification_body, 'notification_body': default_notification_body,
'notification_format': default_notification_format, 'notification_format': default_notification_format,
'schema_version' : 0, 'schema_version' : 0,
'shared_diff_access': False,
'webdriver_delay': None # Extra delay in seconds before extracting text 'webdriver_delay': None # Extra delay in seconds before extracting text
} }
} }

8
changedetectionio/templates/diff.html
Unescape View File

@ -76,8 +76,12 @@
</div> </div>
<div class="tab-pane-inner" id="text"> <div class="tab-pane-inner" id="text">
<div class="tip">Pro-tip: Use <strong>show current snapshot</strong> tab to visualise what will be ignored. <div class="tip">Pro-tip: Use <strong>show current snapshot</strong> tab to visualise what will be ignored.</div>
</div>
{% if password_enabled_and_share_is_off %}
<div class="tip">Pro-tip: You can enable <strong>"share access when password is enabled"</strong> from settings</div>
{% endif %}
<div class="snapshot-age">{{watch_a.snapshot_text_ctime|format_timestamp_timeago}}</div> <div class="snapshot-age">{{watch_a.snapshot_text_ctime|format_timestamp_timeago}}</div>
<table> <table>

5
changedetectionio/templates/settings.html
Unescape View File

@ -57,6 +57,11 @@
{% endif %} {% endif %}
</div> </div>
<div class="pure-control-group">
{{ render_checkbox_field(form.application.form.shared_diff_access, class="shared_diff_access") }}
<span class="pure-form-message-inline">Allow access to view watch diff page when password is enabled (Good for sharing the diff page)
</span>
</div>
<div class="pure-control-group"> <div class="pure-control-group">
{{ render_field(form.application.form.base_url, placeholder="http://yoursite.com:5000/", {{ render_field(form.application.form.base_url, placeholder="http://yoursite.com:5000/",
class="m-d") }} class="m-d") }}

52
changedetectionio/tests/test_access_control.py
Unescape View File

@ -1,18 +1,34 @@
from . util import live_server_setup, extract_UUID_from_client
from flask import url_for from flask import url_for
from . util import live_server_setup import time
def test_check_access_control(app, client): def test_check_access_control(app, client, live_server):
# Still doesnt work, but this is closer. # Still doesnt work, but this is closer.
live_server_setup(live_server)
with app.test_client(use_cookies=True) as c: with app.test_client(use_cookies=True) as c:
# Check we don't have any password protection enabled yet. # Check we don't have any password protection enabled yet.
res = c.get(url_for("settings_page")) res = c.get(url_for("settings_page"))
assert b"Remove password" not in res.data assert b"Remove password" not in res.data
# Enable password check. # add something that we can hit via diff page later
res = c.post(
url_for("import_page"),
data={"urls": url_for('test_random_content_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
time.sleep(2)
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
assert b'1 watches queued for rechecking.' in res.data
time.sleep(2)
# Enable password check and diff page access bypass
res = c.post( res = c.post(
url_for("settings_page"), url_for("settings_page"),
data={"application-password": "foobar", data={"application-password": "foobar",
"application-shared_diff_access": "True",
"requests-time_between_check-minutes": 180, "requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"}, 'application-fetch_backend': "html_requests"},
follow_redirects=True follow_redirects=True
@ -22,9 +38,15 @@ def test_check_access_control(app, client):
# Check we hit the login # Check we hit the login
res = c.get(url_for("index"), follow_redirects=True) res = c.get(url_for("index"), follow_redirects=True)
# Should be logged out
assert b"Login" in res.data assert b"Login" in res.data
# The diff page should return something valid when logged out
res = client.get(url_for("diff_history_page", uuid="first"))
assert b'Random content' in res.data
# Menu should not be available yet # Menu should not be available yet
# assert b"SETTINGS" not in res.data # assert b"SETTINGS" not in res.data
# assert b"BACKUP" not in res.data # assert b"BACKUP" not in res.data
@ -109,3 +131,25 @@ def test_check_access_control(app, client):
assert b"Password protection enabled" not in res.data assert b"Password protection enabled" not in res.data
# Now checking the diff access
# Enable password check and diff page access bypass
res = c.post(
url_for("settings_page"),
data={"application-password": "foobar",
# Should be disabled
# "application-shared_diff_access": "True",
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Password protection enabled." in res.data
# Check we hit the login
res = c.get(url_for("index"), follow_redirects=True)
# Should be logged out
assert b"Login" in res.data
# The diff page should return something valid when logged out
res = client.get(url_for("diff_history_page", uuid="first"))
assert b'Random content' not in res.data

5
changedetectionio/tests/test_backend.py
Unescape View File

@ -3,7 +3,7 @@
import time import time
from flask import url_for from flask import url_for
from urllib.request import urlopen from urllib.request import urlopen
from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks, extract_rss_token_from_UI
sleep_time_for_fetch_thread = 3 sleep_time_for_fetch_thread = 3
@ -76,7 +76,8 @@ def test_check_basic_change_detection_functionality(client, live_server):
assert b'unviewed' in res.data assert b'unviewed' in res.data
# #75, and it should be in the RSS feed # #75, and it should be in the RSS feed
res = client.get(url_for("rss")) rss_token = extract_rss_token_from_UI(client)
res = client.get(url_for("rss", token=rss_token, _external=True))
expected_url = url_for('test_endpoint', _external=True) expected_url = url_for('test_endpoint', _external=True)
assert b'<rss' in res.data assert b'<rss' in res.data

39
changedetectionio/tests/test_rss.py
Unescape View File

@ -0,0 +1,39 @@
#!/usr/bin/python3
import time
from flask import url_for
from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks, extract_rss_token_from_UI
def test_rss_and_token(client, live_server):
set_original_response()
live_server_setup(live_server)
# Add our URL to the import page
res = client.post(
url_for("import_page"),
data={"urls": url_for('test_random_content_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
rss_token = extract_rss_token_from_UI(client)
time.sleep(2)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(2)
# Add our URL to the import page
res = client.get(
url_for("rss", token="bad token", _external=True),
follow_redirects=True
)
assert b"Access denied, bad token" in res.data
res = client.get(
url_for("rss", token=rss_token, _external=True),
follow_redirects=True
)
assert b"Access denied, bad token" not in res.data
assert b"Random content" in res.data

15
changedetectionio/tests/util.py
Unescape View File

@ -70,6 +70,15 @@ def extract_api_key_from_UI(client):
api_key = m.group(1) api_key = m.group(1)
return api_key.strip() return api_key.strip()
# kinda funky, but works for now
def extract_rss_token_from_UI(client):
import re
res = client.get(
url_for("index"),
)
m = re.search('token=(.+?)"', str(res.data))
token_key = m.group(1)
return token_key.strip()
# kinda funky, but works for now # kinda funky, but works for now
def extract_UUID_from_client(client): def extract_UUID_from_client(client):
@ -98,6 +107,12 @@ def wait_for_all_checks(client):
def live_server_setup(live_server): def live_server_setup(live_server):
@live_server.app.route('/test-random-content-endpoint')
def test_random_content_endpoint():
import secrets
return "Random content - {}\n".format(secrets.token_hex(64))
@live_server.app.route('/test-endpoint') @live_server.app.route('/test-endpoint')
def test_endpoint(): def test_endpoint():
ctype = request.args.get('content_type') ctype = request.args.get('content_type')

Write Preview
Loading…
Cancel
Save