From 4cf2d9d7aa79fcc43bdcb5b554e23082aedeb123 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Fri, 20 May 2022 16:00:34 +0200 Subject: [PATCH] WIP on access --- changedetectionio/api/auth.py | 2 +- changedetectionio/forms.py | 3 +- changedetectionio/tests/test_api.py | 83 ++++++++++++++++++++++++----- 3 files changed, 73 insertions(+), 15 deletions(-) diff --git a/changedetectionio/api/auth.py b/changedetectionio/api/auth.py index 67e0662b..1598b449 100644 --- a/changedetectionio/api/auth.py +++ b/changedetectionio/api/auth.py @@ -22,7 +22,7 @@ def check_token(f): if config_api_token_enabled and api_key_header != config_api_token: return make_response( - jsonify("Invalid access - API key invalid.", 403) + jsonify("Invalid access - API key invalid."), 403 ) return f(*args, **kwargs) diff --git a/changedetectionio/forms.py b/changedetectionio/forms.py index 145deca8..4e790c21 100644 --- a/changedetectionio/forms.py +++ b/changedetectionio/forms.py @@ -374,8 +374,7 @@ class globalSettingsApplicationForm(commonSettingsForm): 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) fetch_backend = RadioField('Fetch Method', default="html_requests", choices=content_fetcher.available_fetchers(), validators=[ValidateContentFetcherIsReady()]) - - api_access_token_enabled = BooleanField('API access token security check enabled', default=False) + api_access_token_enabled = BooleanField('API access token security check enabled', default=True, validators=[validators.Optional()]) password = SaltyPasswordField() diff --git a/changedetectionio/tests/test_api.py b/changedetectionio/tests/test_api.py index 99dc42d0..d126c53f 100644 --- a/changedetectionio/tests/test_api.py +++ b/changedetectionio/tests/test_api.py @@ -53,19 +53,35 @@ def is_valid_uuid(val): return False +# kinda funky, but works for now +def _extract_api_key_from_UI(client): + import re + res = client.get( + url_for("settings_page"), + ) + # {{api_key}} + + m = re.search('(.+?)', str(res.data)) + api_key = m.group(1) + return api_key.strip() + + def test_api_simple(client, live_server): live_server_setup(live_server) + api_key = _extract_api_key_from_UI(client) + # Create a watch set_original_response() watch_uuid = None # Validate bad URL - test_url = url_for('test_endpoint', _external=True) + test_url = url_for('test_endpoint', _external=True, + headers={'x-api-key': api_key}, ) res = client.post( url_for("createwatch"), data=json.dumps({"url": "h://xxxxxxxxxom"}), - headers={'content-type': 'application/json'}, + headers={'content-type': 'application/json', 'x-api-key': api_key}, follow_redirects=True ) assert res.status_code == 400 @@ -74,7 +90,7 @@ def test_api_simple(client, live_server): res = client.post( url_for("createwatch"), data=json.dumps({"url": test_url, 'tag': "One, Two", "title": "My test URL"}), - headers={'content-type': 'application/json'}, + headers={'content-type': 'application/json', 'x-api-key': api_key}, follow_redirects=True ) s = json.loads(res.data) @@ -86,7 +102,8 @@ def test_api_simple(client, live_server): # Verify its in the list and that recheck worked res = client.get( - url_for("createwatch") + url_for("createwatch"), + headers={'x-api-key': api_key} ) assert watch_uuid in json.loads(res.data).keys() before_recheck_info = json.loads(res.data)[watch_uuid] @@ -96,13 +113,15 @@ def test_api_simple(client, live_server): set_modified_response() # Trigger recheck of all ?recheck_all=1 client.get( - url_for("createwatch", recheck_all='1') + url_for("createwatch", recheck_all='1'), + headers={'x-api-key': api_key}, ) time.sleep(3) # Did the recheck fire? res = client.get( - url_for("createwatch") + url_for("createwatch"), + headers={'x-api-key': api_key}, ) after_recheck_info = json.loads(res.data)[watch_uuid] assert after_recheck_info['last_checked'] != before_recheck_info['last_checked'] @@ -110,26 +129,30 @@ def test_api_simple(client, live_server): # Check history index list res = client.get( - url_for("watchhistory", uuid=watch_uuid) + url_for("watchhistory", uuid=watch_uuid), + headers={'x-api-key': api_key}, ) history = json.loads(res.data) assert len(history) == 2, "Should have two history entries (the original and the changed)" # Fetch a snapshot by timestamp, check the right one was found res = client.get( - url_for("watchsinglehistory", uuid=watch_uuid, timestamp=list(history.keys())[-1]) + url_for("watchsinglehistory", uuid=watch_uuid, timestamp=list(history.keys())[-1]), + headers={'x-api-key': api_key}, ) assert b'which has this one new line' in res.data # Fetch a snapshot by 'latest'', check the right one was found res = client.get( - url_for("watchsinglehistory", uuid=watch_uuid, timestamp='latest') + url_for("watchsinglehistory", uuid=watch_uuid, timestamp='latest'), + headers={'x-api-key': api_key}, ) assert b'which has this one new line' in res.data # Fetch the whole watch res = client.get( - url_for("watch", uuid=watch_uuid) + url_for("watch", uuid=watch_uuid), + headers={'x-api-key': api_key} ) watch = json.loads(res.data) # @todo how to handle None/default global values? @@ -137,13 +160,49 @@ def test_api_simple(client, live_server): # Finally delete the watch res = client.delete( - url_for("watch", uuid=watch_uuid) + url_for("watch", uuid=watch_uuid), + headers={'x-api-key': api_key}, ) assert res.status_code == 204 # Check via a relist res = client.get( - url_for("createwatch") + url_for("createwatch"), + headers={'x-api-key': api_key} ) watch_list = json.loads(res.data) assert len(watch_list) == 0, "Watch list should be empty" + + +def test_access_denied(client, live_server): + # `config_api_token_enabled` Should be On by default + res = client.get( + url_for("createwatch") + ) + assert res.status_code == 403 + + res = client.get( + url_for("createwatch"), + headers={'x-api-key': "something horrible"} + ) + assert res.status_code == 403 + + # Disable config_api_token_enabled and it should work + res = client.post( + url_for("settings_page"), + data={ + "requests-time_between_check-minutes": 180, + "application-fetch_backend": "html_requests", + "application-api_access_token_enabled": "" + }, + follow_redirects=True + ) + +# with open('/tmp/f.html', 'wb') as f: +# f.write(res.data) + assert b"Settings updated." in res.data + + res = client.get( + url_for("createwatch") + ) + assert res.status_code == 200