UI - Fixing lots of actions that should be a POST style action which led to a 404

Also solves some of GHSA-56fq-63vj-9992 Add-watch-UI should be POST/CSRF protected
This commit is contained in:
dgtlmoon
2026-09-04 12:38:20 +02:00
committed by GitHub
parent e0fb224d41
commit 3a71777499
30 changed files with 381 additions and 70 deletions
@@ -42,7 +42,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
system_default_browser=browser_config.system_default_description(datastore),
)
@add_watch_ui_blueprint.route("/snapshot", methods=['GET'])
@add_watch_ui_blueprint.route("/snapshot", methods=['POST'])
@login_optionally_required
def add_watch_ui_snapshot():
"""One-shot live fetch of an arbitrary URL for the Add Watch visual selector.
@@ -52,6 +52,10 @@ def construct_blueprint(datastore: ChangeDetectionStore):
connect, "Goto site", grab the screenshot + xpath element data, then tear
the browser down again. Element selection then happens client-side on the
returned data, exactly like the watch Edit page's visual selector.
POST-only and CSRF protected on purpose: this drives a real browser fetch and
writes a temporary watch dir, so as a GET it could be triggered cross-origin
(or by any tag/link that issues a GET) without the operator's consent.
"""
import base64
from changedetectionio.blueprint.browser_steps import (
@@ -71,7 +75,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# backslash/parser-differential rejection of GHSA-rph4-96w6-q594 (GHSA-56fq-63vj-9992).
# Note this fetch never reaches difference_detection_processor.call_browser(), so it gets
# no gating from there - it has to validate for itself.
url = (request.args.get('url') or '').strip()
url = (request.form.get('url') or '').strip()
ok, reason = is_fetch_url_allowed(url)
if not ok:
logger.warning(f"Add-watch snapshot: refused '{url}' - {reason}")
@@ -82,7 +86,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# Either way it has to be able to render a preview - the plain HTTP client
# produces no screenshot and no element data, so previewing with it is pointless
# (and it used to be the silent default here, see the system-default bug).
fetcher_name = (request.args.get('fetch_backend') or '').strip() or browser_config.default_visual_browser(datastore)
fetcher_name = (request.form.get('fetch_backend') or '').strip() or browser_config.default_visual_browser(datastore)
if not fetcher_name or not browser_config.is_visual_capable(fetcher_name, datastore):
logger.warning(f"Add-watch snapshot: refused browser '{fetcher_name}' for '{url}'")
return make_response('No interactive browser available that can render a live preview '
@@ -59,9 +59,18 @@ $(document).ready(() => {
$.ajax({
url: add_watch_snapshot_url,
// POST, never GET - this makes the server-side browser fetch a URL of our
// choosing, so it must not be triggerable cross-origin. csrf.js adds the
// X-CSRFToken header to every non-GET ajax call; the CSRF field on the form
// is sent too so it works even if that handler hasn't run yet.
method: 'POST',
// Preview with the browser picked in the list - that same browser is what
// gets saved on the watch, so what you see here is what it will check with.
data: {url: url, fetch_backend: $('input[name="fetch_backend"]:checked').val() || ''},
data: {
url: url,
fetch_backend: $('input[name="fetch_backend"]:checked').val() || '',
csrf_token: $('#new-watch-form input[name="csrf_token"]').val() || '',
},
dataType: 'json',
}).done((data) => {
showState('ready');
@@ -98,7 +98,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
backups_blueprint.register_blueprint(construct_restore_blueprint(datastore))
backup_threads = []
@backups_blueprint.route("/request-backup", methods=['GET'])
@backups_blueprint.route("/request-backup", methods=['POST'])
@login_optionally_required
def request_backup():
if any(thread.is_alive() for thread in backup_threads):
@@ -35,8 +35,10 @@
</p>
{% endif %}
<a class="pure-button pure-button-primary"
href="{{ url_for('backups.request_backup') }}">{{ _('Create backup') }}</a>
<form method="POST" action="{{ url_for('backups.request_backup') }}" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="pure-button pure-button-primary">{{ _('Create backup') }}</button>
</form>
{% if available_backups %}
{# POST + CSRF token: this permanently deletes every backup archive, so it must
not be reachable from a bare GET (an <img src=...> on any page the operator
@@ -296,7 +296,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return browsersteps_start_session
@browser_steps_blueprint.route("/browsersteps_start_session", methods=['GET'])
@browser_steps_blueprint.route("/browsersteps_start_session", methods=['POST'])
@login_optionally_required
def browsersteps_start_session():
# A new session was requested, return sessionID
@@ -100,7 +100,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
results = _recalc_check_status(uuid=uuid)
return results
@check_proxies_blueprint.route("/<uuid_str:uuid>/start", methods=['GET'])
@check_proxies_blueprint.route("/<uuid_str:uuid>/start", methods=['POST'])
@login_optionally_required
def start_check(uuid):
+1 -1
View File
@@ -130,7 +130,7 @@ def construct_llm_blueprint(datastore: ChangeDetectionStore):
logger.exception("LLM model list full traceback:")
return jsonify({'models': [], 'error': str(e)}), 400
@llm_blueprint.route("/test", methods=['GET'])
@llm_blueprint.route("/test", methods=['POST'])
@login_optionally_required
def llm_test():
from flask import request
@@ -577,7 +577,10 @@
if (mult.trim()) params.set('local_token_multiplier', mult.trim());
try {
const resp = await fetch('{{ url_for("settings.llm.llm_test") }}?' + params);
const resp = await fetch('{{ url_for("settings.llm.llm_test") }}?' + params, {
method: 'POST',
headers: {'X-CSRFToken': csrftoken}
});
const data = await resp.json();
if (data.ok) {
result.style.cssText = 'display:block; background:rgba(39,174,96,0.08); border:1px solid rgba(39,174,96,0.3); border-radius:5px; padding:0.6em 0.85em; font-size:0.88em; line-height:1.45;';
+1 -1
View File
@@ -455,7 +455,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
return redirect(url_for('watchlist.index'))
@ui_blueprint.route("/language/auto-detect", methods=['GET'])
@ui_blueprint.route("/language/auto-detect", methods=['POST'])
def delete_locale_language_session_var_if_it_exists():
"""Clear the session locale preference to auto-detect from browser Accept-Language header"""
if 'locale' in session:
+2 -2
View File
@@ -836,7 +836,7 @@ def changedetection_app(config=None, datastore_o=None):
# Pass the current request path so users are redirected back after login
return redirect(url_for('login', redirect=request.path))
@app.route('/logout')
@app.route('/logout', methods=['POST'])
def logout():
flask_login.logout_user()
@@ -850,7 +850,7 @@ def changedetection_app(config=None, datastore_o=None):
# Otherwise just go to watchlist
return redirect(url_for('watchlist.index'))
@app.route('/set-language/<locale>')
@app.route('/set-language/<locale>', methods=['POST'])
def set_language(locale):
"""Set the user's preferred language in the session"""
if not request.cookies:
+1 -1
View File
@@ -284,7 +284,7 @@ $(document).ready(function () {
$('#browser-steps-ui .loader .spinner').show();
// Request a new session
$.ajax({
type: "GET",
type: "POST",
url: browser_steps_start_url,
statusCode: {
400: function () {
+1 -1
View File
@@ -73,7 +73,7 @@ $(function () {
// Request start, needs CSRF?
$.ajax({
type: "GET",
type: "POST",
url: recheck_proxy_start_url,
}).done(function (data) {
$.each(data, function (proxy_key, state) {
@@ -122,7 +122,7 @@
li {
border-bottom: 1px solid var(--color-border-table-cell);
>* {
>*, >form>button {
display: block;
padding: 1rem 1.5rem;
color: var(--color-text);
@@ -134,6 +134,14 @@
background: var(--color-background-menu-link-hover);
}
}
// Buttons shrink-wrap and centre their label; anchors don't. No global
// border-box reset here, so this must not be folded into the rule above.
>form>button {
box-sizing: border-box;
width: 100%;
text-align: left;
}
&#menu-pause, &#menu-mute {
display: none;
}
@@ -18,10 +18,20 @@
flex-direction: column;
gap: 0.5rem;
padding: 0.5rem 0;
// The csrf mini-form around each option is layout-transparent, so the buttons
// stay the flex items.
> form {
display: contents;
}
}
.language-option {
display: flex;
background: none;
font: inherit;
text-align: left;
cursor: pointer;
align-items: center;
gap: 1rem;
padding: 0.25rem;
@@ -15,6 +15,13 @@
.pure-menu-item {
height: initial;
// Mini POST forms (pause/mute/log out need a csrf_token) are layout-transparent,
// so the button inside sits where the plain <a> used to.
> form {
display: contents;
}
svg {
height: 1.2rem;
}
File diff suppressed because one or more lines are too long
+10 -4
View File
@@ -271,13 +271,19 @@
<div class="modal-body">
<div class="language-list">
{% for locale, lang_data in available_languages.items()|sort %}
<a href="{{ url_for('set_language', locale=locale, redirect=request.path) }}" class="language-option" data-locale="{{ locale }}">
<span class="lang-option {{ lang_data.flag }}"></span> <span class="language-name">{{ lang_data.name }}</span>
</a>
<form method="POST" action="{{ url_for('set_language', locale=locale, redirect=request.path) }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="language-option" data-locale="{{ locale }}">
<span class="lang-option {{ lang_data.flag }}"></span> <span class="language-name">{{ lang_data.name }}</span>
</button>
</form>
{% endfor %}
</div>
<div>
<a href="{{ url_for('ui.delete_locale_language_session_var_if_it_exists', redirect=request.path) }}" >{{ _('Auto-detect from browser') }}</a>
<form method="POST" action="{{ url_for('ui.delete_locale_language_session_var_if_it_exists', redirect=request.path) }}" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bare-btn">{{ _('Auto-detect from browser') }}</button>
</form>
</div>
<div>
{{ _('Language support is in beta, please help us improve by opening a PR on GitHub with any updates.') }}
+4 -1
View File
@@ -18,7 +18,10 @@
</li>
{%- if current_user.is_authenticated -%}
<li class="pure-menu-item menu-collapsible">
<a href="{{ url_for('logout', redirect=request.path) }}" ><i data-feather="log-out" class="action-icon"></i>&nbsp;{{ _('Log out') }}</a>
<form method="POST" action="{{ url_for('logout', redirect=request.path) }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bare-btn"><i data-feather="log-out" class="action-icon"></i>&nbsp;{{ _('Log out') }}</button>
</form>
</li>
{%- endif -%}
@@ -84,10 +84,12 @@ def test_socks5(client, live_server, measure_memory_usage, datastore_path):
# PROXY CHECKER WIDGET CHECK - this needs more checking
uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))
res = client.get(
# POST only - it kicks off real fetches through every configured proxy
res = client.post(
url_for("check_proxies.start_check", uuid=uuid),
follow_redirects=True
)
assert res.status_code == 200
# It's probably already finished super fast :(
#assert b"RUNNING" in res.data
@@ -125,7 +125,7 @@ def test_check_access_control(app, client, live_server, measure_memory_usage, da
follow_redirects=True
)
res = c.get(url_for("logout"),
res = c.post(url_for("logout"),
follow_redirects=True)
assert b"Login" in res.data
@@ -79,26 +79,41 @@ def test_snapshot_refuses_browser_that_cannot_preview(client, live_server, measu
from changedetectionio.blueprint.add_watch_ui import browser_config
monkeypatch.setattr(browser_config, 'is_visual_capable', lambda name, datastore: False)
snapshot_url = url_for('add_watch_ui.add_watch_ui_snapshot')
# Nothing capable, and no explicit browser asked for -> nothing to preview with
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com'))
res = client.post(snapshot_url, data={'url': 'https://example.com'})
assert res.status_code == 400
assert b'No interactive browser' in res.data
# Explicitly asking for a browser that can't preview is refused just the same
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='html_requests'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': 'html_requests'})
assert res.status_code == 400
# A made-up name never resolves to a capable fetcher either (real capability lookup here)
monkeypatch.undo()
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='../../etc/passwd'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': '../../etc/passwd'})
assert res.status_code == 400
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='os'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': 'os'})
assert res.status_code == 400
def test_snapshot_is_post_only(client, live_server, measure_memory_usage, datastore_path):
"""A GET must not reach the endpoint at all.
/snapshot drives a real server-side browser fetch and hands the rendered result back in
the response (GHSA-56fq-63vj-9992). As a GET that is reachable by anything that can make
the operator's browser issue a request - an <img>/<iframe>/link from another site - with
no CSRF token in play. POST-only + CSRFProtect means only our own page can trigger it.
"""
# Method mismatch surfaces as 404 here rather than 405
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot') + '?url=https://example.com')
assert res.status_code in (404, 405)
def test_submit_rejects_unknown_fetcher(client, live_server, measure_memory_usage, datastore_path):
"""A posted browser is checked server side, so a doctored form can't pin a junk fetcher."""
datastore = _datastore(client)
+2 -2
View File
@@ -24,7 +24,7 @@ def test_backup(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
# Launch the thread in the background to create the backup
res = client.get(
res = client.post(
url_for("backups.request_backup"),
follow_redirects=True
)
@@ -136,7 +136,7 @@ def test_backup_restore(client, live_server, measure_memory_usage, datastore_pat
wait_for_all_checks(client)
# Create a full backup
client.get(url_for("backups.request_backup"), follow_redirects=True)
client.post(url_for("backups.request_backup"), follow_redirects=True)
time.sleep(4)
# Download the latest backup zip
+21 -21
View File
@@ -11,7 +11,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
# Be sure we got a session cookie
res = client.get(url_for("watchlist.index"), follow_redirects=True)
res = client.get(
res = client.post(
url_for("set_language", locale="zh_Hant_TW"), # Traditional
follow_redirects=True
)
@@ -21,7 +21,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
assert '選擇語言'.encode() in res.data
# Check second set works
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -30,7 +30,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
assert b"Select Language" in res.data, "Second set of language worked"
# Check arbitration between zh_Hant_TW<->zh
res = client.get(
res = client.post(
url_for("set_language", locale="zh"), # Simplified chinese
follow_redirects=True
)
@@ -89,7 +89,7 @@ def test_language_switching(client, live_server, measure_memory_usage, datastore
client.get(url_for("add_watch_ui.add_watch_ui_index"), follow_redirects=True)
# Step 1: Set the language to Italian using the /set-language endpoint
res = client.get(
res = client.post(
url_for("set_language", locale="it"),
follow_redirects=True
)
@@ -119,7 +119,7 @@ def test_language_switching(client, live_server, measure_memory_usage, datastore
# NB: use 'en_GB' not 'en' — only the variants are in language_codes; the
# plain 'en' code is silently rejected by set_language and the locale would
# remain at 'it', defeating the round-trip assertion below.
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -152,7 +152,7 @@ def test_invalid_locale(client, live_server, measure_memory_usage, datastore_pat
# bare 'en' is NOT in language_codes and is silently rejected by
# set_language, so passing it here would leave the session locale unset
# and let the (unrelated) Accept-Language fallback decide what renders.
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -160,7 +160,7 @@ def test_invalid_locale(client, live_server, measure_memory_usage, datastore_pat
assert res.status_code == 200
# Try to set an invalid locale
res = client.get(
res = client.post(
url_for("set_language", locale="invalid_locale_xyz"),
follow_redirects=True
)
@@ -190,7 +190,7 @@ def test_language_persistence_in_session(client, live_server, measure_memory_usa
client.get(url_for("watchlist.index"), follow_redirects=True)
# Set language to Italian
res = client.get(
res = client.post(
url_for("set_language", locale="it"),
follow_redirects=True
)
@@ -215,7 +215,7 @@ def test_language_persistence_in_session(client, live_server, measure_memory_usa
assert sess.get('locale') == 'it', "Locale should be set in session"
# Call auto-detect to clear the locale
res = client.get(
res = client.post(
url_for("ui.delete_locale_language_session_var_if_it_exists"),
follow_redirects=True
)
@@ -254,7 +254,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
client.get(url_for("watchlist.index"), follow_redirects=True)
# Set language with a redirect parameter (simulating language change from /settings)
res = client.get(
res = client.post(
url_for("set_language", locale="de", redirect="/settings"),
follow_redirects=False
)
@@ -268,7 +268,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
assert sess.get('locale') == 'de'
# Test with invalid locale (should still redirect safely)
res = client.get(
res = client.post(
url_for("set_language", locale="invalid_locale", redirect="/settings"),
follow_redirects=False
)
@@ -276,7 +276,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
assert '/settings' in res.location
# Test with malicious redirect (should default to watchlist)
res = client.get(
res = client.post(
url_for("set_language", locale="en", redirect="https://evil.com"),
follow_redirects=False
)
@@ -296,7 +296,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
client.get(url_for("watchlist.index"), follow_redirects=True)
# Test Italian translations
res = client.get(url_for("set_language", locale="it"), follow_redirects=True)
res = client.post(url_for("set_language", locale="it"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -312,7 +312,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Korean translations
res = client.get(url_for("set_language", locale="ko"), follow_redirects=True)
res = client.post(url_for("set_language", locale="ko"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -332,7 +332,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Chinese Simplified translations
res = client.get(url_for("set_language", locale="zh"), follow_redirects=True)
res = client.post(url_for("set_language", locale="zh"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -348,7 +348,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test German translations
res = client.get(url_for("set_language", locale="de"), follow_redirects=True)
res = client.post(url_for("set_language", locale="de"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -363,7 +363,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Russian translations
res = client.get(url_for("set_language", locale="ru"), follow_redirects=True)
res = client.post(url_for("set_language", locale="ru"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -378,7 +378,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Traditional Chinese (zh_Hant_TW) translations
res = client.get(url_for("set_language", locale="zh_Hant_TW"), follow_redirects=True)
res = client.post(url_for("set_language", locale="zh_Hant_TW"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -627,7 +627,7 @@ def test_session_locale_overrides_accept_language(client, live_server, measure_m
"Expected Taiwan flag 'fi fi-tw' from auto-detect"
# Step 2: User explicitly selects Korean language
res = client.get(
res = client.post(
url_for("set_language", locale="ko"),
headers={'Accept-Language': 'zh-TW,zh;q=0.9,en;q=0.8'}, # Browser still sends zh-TW
follow_redirects=True
@@ -700,7 +700,7 @@ def test_clear_history_translated_confirmation(client, live_server, measure_memo
wait_for_all_checks(client)
# Set language to German
res = client.get(
res = client.post(
url_for("set_language", locale="de"),
follow_redirects=True
)
@@ -726,7 +726,7 @@ def test_clear_history_translated_confirmation(client, live_server, measure_memo
"German confirmation word 'loschen' should be accepted (issue #3865)"
# Switch back to English and verify English word still works
res = client.get(
res = client.post(
url_for("set_language", locale="en_US"),
follow_redirects=True
)
@@ -25,7 +25,7 @@ def test_language_endpoints_work_for_anonymous_users(client, live_server, measur
follow_redirects=True)
assert res.status_code == 200
client.get(url_for("logout"), follow_redirects=True)
client.post(url_for("logout"), follow_redirects=True)
# Both language links are rendered on the login page, so both must be reachable
res = client.get(url_for("login"))
@@ -33,13 +33,13 @@ def test_language_endpoints_work_for_anonymous_users(client, live_server, measur
assert b'language-selector' in res.data, "Language modal trigger should render for anonymous users"
# Picking a specific language must not redirect to the login page
res = client.get(url_for("set_language", locale="de"), follow_redirects=False)
res = client.post(url_for("set_language", locale="de"), follow_redirects=False)
assert res.status_code == 302
assert '/login' not in res.headers.get("Location", ""), \
"set_language must not bounce anonymous users to /login"
# ...and neither must clearing it back to auto-detect
res = client.get(url_for("ui.delete_locale_language_session_var_if_it_exists"), follow_redirects=False)
res = client.post(url_for("ui.delete_locale_language_session_var_if_it_exists"), follow_redirects=False)
assert res.status_code == 302
assert '/login' not in res.headers.get("Location", ""), \
"Auto-detect must not bounce anonymous users to /login (it renders on the login page)"
@@ -393,12 +393,12 @@ def test_llm_models_endpoint_blocks_private_api_base(
def test_llm_test_endpoint_blocks_private_api_base(
client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""GET /settings/llm/test must refuse api_base pointing at private/loopback
"""POST /settings/llm/test must refuse api_base pointing at private/loopback
hosts and must never reach litellm.completion()."""
monkeypatch.delenv('ALLOW_IANA_RESTRICTED_ADDRESSES', raising=False)
for bad in _SSRF_PRIVATE_HOSTS:
res = client.get(
res = client.post(
url_for('settings.llm.llm_test'),
query_string={'model': 'openai/gpt-4', 'api_base': bad},
)
@@ -530,7 +530,7 @@ def test_llm_test_refuses_to_leak_stored_key_to_different_api_base(
monkeypatch.setattr(llm_client, 'completion',
lambda **kw: calls.append(kw) or ('', 0, 0, 0))
res = client.get(
res = client.post(
url_for('settings.llm.llm_test'),
query_string={
'model': 'gpt-4o-mini',
@@ -38,7 +38,7 @@ def test_rss_tag_feed_ignores_security_token(client, live_server, datastore_path
wait_for_all_checks(client)
# Logout
client.get(url_for("logout"), follow_redirects=True)
client.post(url_for("logout"), follow_redirects=True)
# Request the tag RSS feed WITH the token
res = client.get(
+5 -5
View File
@@ -440,7 +440,7 @@ def test_login_redirect_with_password(client, live_server, measure_memory_usage,
assert b"evil.com" not in res.data
# Logout for cleanup
client.get(url_for("logout"))
client.post(url_for("logout"))
# Test 5: Incorrect password with redirect should stay on login page
res = client.post(
@@ -483,7 +483,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
client.application.config['DATASTORE'].data['settings']['application']['password'] = salted_pass
# Logout to ensure we're not authenticated
client.get(url_for("logout"))
client.post(url_for("logout"))
# Try to access a protected page (edit page for first watch)
res = client.get(
@@ -524,7 +524,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
assert b'Edit' in res.data or b'Watching' in res.data
# Cleanup
client.get(url_for("logout"))
client.post(url_for("logout"))
del client.application.config['DATASTORE'].data['settings']['application']['password']
@@ -554,7 +554,7 @@ def test_logout_with_redirect(client, live_server, measure_memory_usage, datasto
assert res.status_code == 200
# Now logout with a redirect parameter (simulating logout from /settings)
res = client.get(
res = client.post(
url_for("logout", redirect="/settings"),
follow_redirects=False
)
@@ -961,7 +961,7 @@ def test_ghsa_8757_69j2_hx56_backup_restore_history_path_traversal(client, live_
wait_for_all_checks(client)
# Download a legitimate backup to use as a template
client.get(url_for("backups.request_backup"), follow_redirects=True)
client.post(url_for("backups.request_backup"), follow_redirects=True)
time.sleep(4)
res = client.get(url_for("backups.download_backup", filename="latest"), follow_redirects=True)
assert res.content_type == "application/zip"
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""Tests for the shared "may the server fetch this URL?" gate.
# run from dir above changedetectionio/ dir
# python3 -m unittest changedetectionio.tests.unit.test_fetch_url_gate
Every server-side fetch entry point routes through validate_url.is_fetch_url_allowed(). Before it
existed, the file:// and private-IP rules were enforced inline in call_browser() only, so any fetch
path that did not go through call_browser() was unprotected:
* a "Goto URL" browser step could read file:///etc/passwd (GHSA-hm22-wg2m-35v4)
* /add-watch-ui/snapshot url= could fetch internal hosts (GHSA-56fq-63vj-9992)
These tests pin the gate's rules AND the browser-step choke point, so a future fetch path that
forgets to call the gate is the only way to regress it.
"""
import asyncio
import unittest
from unittest.mock import patch
from changedetectionio.browser_steps.browser_steps import steppable_browser_interface
from changedetectionio.validate_url import (
is_fetch_url_allowed,
is_special_purpose_ip,
validate_fetch_url,
validate_fetch_url_async,
)
# tests/conftest.py sets ALLOW_IANA_RESTRICTED_ADDRESSES=true for the functional suite, so the
# locked-down default has to be re-asserted explicitly rather than assumed.
LOCKED_DOWN = {'ALLOW_IANA_RESTRICTED_ADDRESSES': 'false', 'ALLOW_FILE_URI': 'false'}
OPTED_IN = {'ALLOW_IANA_RESTRICTED_ADDRESSES': 'true', 'ALLOW_FILE_URI': 'true'}
class TestFetchUrlGate(unittest.TestCase):
def assertBlocked(self, url):
ok, reason = is_fetch_url_allowed(url)
self.assertFalse(ok, f"URL '{url}' should have been blocked")
self.assertTrue(reason, f"URL '{url}' was blocked without a reason to show the user")
def assertAllowed(self, url):
ok, reason = is_fetch_url_allowed(url)
self.assertTrue(ok, f"URL '{url}' should have been allowed, got: {reason}")
def test_file_uri_blocked_by_default(self):
with patch.dict('os.environ', LOCKED_DOWN):
# All the spellings that reach the same local file
for url in ('file:///etc/passwd', 'FILE:///etc/passwd', 'file:/etc/passwd', 'file://etc/passwd'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_file_uri_allowed_when_operator_opts_in(self):
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('file:///etc/passwd')
def test_file_uri_blocked_even_if_safe_protocol_regex_was_loosened(self):
"""An operator who widens SAFE_PROTOCOL_REGEX for some other scheme must not get local
file reads thrown in for free - hence the explicit file: check ahead of is_safe_valid_url()."""
env = dict(LOCKED_DOWN, SAFE_PROTOCOL_REGEX='^(http|https|ftp|file):')
with patch.dict('os.environ', env):
self.assertBlocked('file:///etc/passwd')
def test_private_and_reserved_addresses_blocked_by_default(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://127.0.0.1:5000/',
'http://localhost/',
'http://169.254.169.254/latest/meta-data/', # cloud metadata
'http://192.168.1.1/',
'http://10.0.0.1/',
'http://[::1]/'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_cgnat_and_other_non_global_addresses_blocked_by_default(self):
"""GHSA-gwph-fp79-379w - the 0.54.1 predicate only tested is_private/is_loopback/
is_link_local/is_reserved, none of which are True for RFC 6598 CGNAT space, so
100.64.0.0/10 (an ISP's other subscribers, CPE admin panels, CGNAT gateways) stayed
fetchable. These are IP literals, so no DNS is involved and CI cannot flake."""
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://100.64.0.1/', # RFC 6598 CGNAT, first usable
'http://100.127.255.254/', # RFC 6598 CGNAT, last usable
'http://100.100.100.100/', # inside CGNAT (Alibaba Cloud metadata)
'http://192.88.99.1/', # RFC 7526 deprecated 6to4 relay anycast
'http://224.0.0.1/', # IPv4 multicast all-hosts
'http://[ff02::1]/'): # IPv6 multicast all-nodes
with self.subTest(url=url):
self.assertBlocked(url)
def test_cgnat_allowed_when_operator_opts_in(self):
"""CGNAT is legitimate for operators monitoring their own carrier network, so the
opt-in has to release it the same way it releases 127.0.0.1."""
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('http://100.64.0.1/')
def test_special_purpose_ip_classification(self):
"""The predicate itself, without DNS - one place to pin what is and is not fetchable."""
for ip in ('100.64.0.1', '100.127.255.254', '192.88.99.1', '224.0.0.1', 'ff02::1',
'127.0.0.1', '10.0.0.1', '169.254.169.254', '192.168.1.1', '::1',
'0.0.0.0', '255.255.255.255', '198.18.0.1', 'fc00::1', 'fe80::1',
'::ffff:100.64.0.1', # CGNAT wrapped as an IPv4-mapped IPv6 address
'2002:6440:1::'): # CGNAT wrapped as a 6to4 address
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertTrue(blocked, f"{ip} should be refused")
self.assertTrue(why, f"{ip} was refused without a stated reason")
for ip in ('1.1.1.1', '8.8.8.8', '93.184.216.34', '2606:4700:4700::1111'):
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertFalse(blocked, f"public address {ip} was refused as '{why}'")
def test_cgnat_boundaries_are_exact(self):
"""100.64.0.0/10 ends at 100.127.255.255 - 100.63.x and 100.128.x are ordinary public
space and must not be collateral damage from a /8-sized over-block."""
for ip in ('100.63.255.255', '100.128.0.0'):
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertFalse(blocked, f"public address {ip} was refused as '{why}'")
def test_private_addresses_allowed_when_operator_opts_in(self):
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('http://127.0.0.1:5000/')
def test_source_prefix_is_stripped_before_the_hostname_check(self):
"""Load-bearing, not cosmetic: urlparse('source:http://127.0.0.1/') reports NO hostname,
so leaving the prefix on would hand the private-IP check nothing to look at and let it pass."""
with patch.dict('os.environ', LOCKED_DOWN):
self.assertBlocked('source:http://127.0.0.1/')
self.assertBlocked('SOURCE:http://169.254.169.254/')
self.assertBlocked('source:file:///etc/passwd')
def test_jinja2_is_rendered_before_the_hostname_check(self):
"""The fetch uses the rendered URL, so the rendered URL is what must be judged - otherwise
a template expression hides the real target from the check."""
with patch.dict('os.environ', LOCKED_DOWN):
self.assertBlocked("http://{{ '127.0.0.1' }}/")
self.assertBlocked("http://{% if 1 %}127.0.0.1{% endif %}/")
def test_parser_differential_payload_always_rejected(self):
"""GHSA-rph4-96w6-q594: urlparse sees PUBLIC, urllib3 connects to INTERNAL. A backslash has
no legitimate use in a URL, so this is refused even with both opt-ins enabled."""
for env in (LOCKED_DOWN, OPTED_IN):
with self.subTest(env=env), patch.dict('os.environ', env):
self.assertBlocked('http://127.0.0.1:8888\\@example.com/')
def test_unsupported_schemes_rejected(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('javascript:alert(1)', 'data:text/html,<h1>x', 'chrome://version'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_empty_input_rejected(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('', ' ', None):
with self.subTest(url=url):
self.assertBlocked(url)
def test_ordinary_public_urls_still_allowed(self):
# Unresolvable hostnames are allowed by design (DNS may be down, domain not yet live), so
# these pass with or without working DNS in CI.
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('https://example.com/',
'source:https://example.com/',
'https://example.com/path?a=b&c=d#frag'):
with self.subTest(url=url):
self.assertAllowed(url)
def test_validate_fetch_url_raises_with_the_reason(self):
with patch.dict('os.environ', LOCKED_DOWN):
with self.assertRaises(ValueError):
validate_fetch_url('file:///etc/passwd')
validate_fetch_url('https://example.com/') # must not raise
def test_validate_fetch_url_async_raises_with_the_reason(self):
with patch.dict('os.environ', LOCKED_DOWN):
with self.assertRaises(ValueError):
asyncio.run(validate_fetch_url_async('http://127.0.0.1/'))
asyncio.run(validate_fetch_url_async('https://example.com/')) # must not raise
class _RecordingPage:
"""Stands in for the Playwright page so we can assert navigation never happened."""
def __init__(self):
self.goto_calls = []
async def goto(self, url, **kwargs):
self.goto_calls.append(url)
return None
async def wait_for_timeout(self, ms):
return None
class TestBrowserStepGotoUrlGate(unittest.TestCase):
"""GHSA-hm22-wg2m-35v4 - browser step values are raw user input and were never validated.
action_goto_url() is the single choke point for every navigation we initiate (the "Goto URL"
step, "Goto site", the live Browser Steps UI and the Add Watch preview all land here), so the
assertion that matters is that page.goto() is never reached for a refused URL.
"""
def _interface(self, start_url='https://example.com/'):
interface = steppable_browser_interface(start_url=start_url)
interface.page = _RecordingPage()
return interface
def test_goto_url_step_cannot_read_local_files(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface()
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_url(value='file:///etc/passwd'))
self.assertEqual(interface.page.goto_calls, [], "Chromium was navigated to a refused URL")
def test_goto_url_step_cannot_reach_private_addresses(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://127.0.0.1:5000/', 'http://169.254.169.254/latest/meta-data/'):
with self.subTest(url=url):
interface = self._interface()
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_url(value=url))
self.assertEqual(interface.page.goto_calls, [])
def test_goto_site_step_validates_the_start_url_too(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface(start_url='source:http://127.0.0.1/')
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_site())
self.assertEqual(interface.page.goto_calls, [])
def test_permitted_url_still_navigates(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface()
asyncio.run(interface.action_goto_url(value='https://example.com/'))
self.assertEqual(interface.page.goto_calls, ['https://example.com/'])
if __name__ == '__main__':
unittest.main()
@@ -205,8 +205,8 @@ def test_browsersteps_edit_UI_startsession(client, live_server, measure_memory_u
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, extras={'fetch_backend': 'html_webdriver', 'paused': True})
# Test starting a browsersteps session
res = client.get(
# Test starting a browsersteps session (POST only - it spins up a real browser)
res = client.post(
url_for("browser_steps.browsersteps_start_session", uuid=uuid),
follow_redirects=True
)
+1 -1
View File
@@ -8,7 +8,7 @@ msgid ""
msgstr ""
"Project-Id-Version: changedetection.io 0.60.2\n"
"Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
"POT-Creation-Date: 2026-09-03 21:31+0200\n"
"POT-Creation-Date: 2026-09-04 10:59+0200\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"