From fe7aa38c651d73fe5f41ce09855fa8f97193747b Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Mon, 23 Feb 2026 21:43:25 +0100 Subject: [PATCH] CVE-2026-27696 - Server-Side Request Forgery (SSRF) via Watch URLs, set env var `ALLOW_IANA_RESTRICTED_ADDRESSES` to `true` to access IANA reserved URLs such as http://169.254.169.254, http://10.0.0.1/, http://127.0.0.1/, etc. --- .../content_fetchers/requests.py | 38 +++++- changedetectionio/run_basic_tests.sh | 4 +- changedetectionio/store/__init__.py | 7 +- changedetectionio/tests/conftest.py | 4 + changedetectionio/tests/test_security.py | 129 ++++++++++++++++++ changedetectionio/validate_url.py | 27 ++++ 6 files changed, 204 insertions(+), 5 deletions(-) diff --git a/changedetectionio/content_fetchers/requests.py b/changedetectionio/content_fetchers/requests.py index 0e37ffacf..87a6cf9ae 100644 --- a/changedetectionio/content_fetchers/requests.py +++ b/changedetectionio/content_fetchers/requests.py @@ -1,4 +1,5 @@ from loguru import logger +from urllib.parse import urljoin, urlparse import hashlib import os import re @@ -7,6 +8,7 @@ import asyncio from changedetectionio import strtobool from changedetectionio.content_fetchers.exceptions import BrowserStepsInUnsupportedFetcher, EmptyReply, Non200ErrorCodeReceived from changedetectionio.content_fetchers.base import Fetcher +from changedetectionio.validate_url import is_private_hostname # "html_requests" is listed as the default fetcher in store.py! @@ -79,14 +81,48 @@ class fetcher(Fetcher): if strtobool(os.getenv('ALLOW_FILE_URI', 'false')) and url.startswith('file://'): from requests_file import FileAdapter session.mount('file://', FileAdapter()) + + allow_iana_restricted = strtobool(os.getenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false')) + try: + # Fresh DNS check at fetch time — catches DNS rebinding regardless of add-time cache. + if not allow_iana_restricted: + parsed_initial = urlparse(url) + if parsed_initial.hostname and is_private_hostname(parsed_initial.hostname): + raise Exception(f"Fetch blocked: '{url}' resolves to a private/reserved IP address. " + f"Set ALLOW_IANA_RESTRICTED_ADDRESSES=true to allow.") + r = session.request(method=request_method, data=request_body.encode('utf-8') if type(request_body) is str else request_body, url=url, headers=request_headers, timeout=timeout, proxies=proxies, - verify=False) + verify=False, + allow_redirects=False) + + # Manually follow redirects so each hop's resolved IP can be validated, + # preventing SSRF via an open redirect on a public host. + current_url = url + for _ in range(10): + if not r.is_redirect: + break + location = r.headers.get('Location', '') + redirect_url = urljoin(current_url, location) + if not allow_iana_restricted: + parsed_redirect = urlparse(redirect_url) + if parsed_redirect.hostname and is_private_hostname(parsed_redirect.hostname): + raise Exception(f"Redirect blocked: '{redirect_url}' resolves to a private/reserved IP address.") + current_url = redirect_url + r = session.request('GET', redirect_url, + headers=request_headers, + timeout=timeout, + proxies=proxies, + verify=False, + allow_redirects=False) + else: + raise Exception("Too many redirects") + except Exception as e: msg = str(e) if proxies and 'SOCKSHTTPSConnectionPool' in msg: diff --git a/changedetectionio/run_basic_tests.sh b/changedetectionio/run_basic_tests.sh index e3ea636ea..784b2a5ba 100755 --- a/changedetectionio/run_basic_tests.sh +++ b/changedetectionio/run_basic_tests.sh @@ -44,12 +44,12 @@ data_sanity_test () { cd .. TMPDIR=$(mktemp -d) PORT_N=$((5000 + RANDOM % (6501 - 5000))) - ./changedetection.py -p $PORT_N -d $TMPDIR -u "https://localhost?test-url-is-sanity=1" & + ALLOW_IANA_RESTRICTED_ADDRESSES=true ./changedetection.py -p $PORT_N -d $TMPDIR -u "https://localhost?test-url-is-sanity=1" & PID=$! sleep 5 kill $PID sleep 2 - ./changedetection.py -p $PORT_N -d $TMPDIR & + ALLOW_IANA_RESTRICTED_ADDRESSES=true ./changedetection.py -p $PORT_N -d $TMPDIR & PID=$! sleep 5 # On a restart the URL should still be there diff --git a/changedetectionio/store/__init__.py b/changedetectionio/store/__init__.py index f5884cfc0..70561c540 100644 --- a/changedetectionio/store/__init__.py +++ b/changedetectionio/store/__init__.py @@ -728,8 +728,11 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): return False if not is_safe_valid_url(url): - flash(gettext('Watch protocol is not permitted or invalid URL format'), 'error') - + from flask import has_request_context + if has_request_context(): + flash(gettext('Watch protocol is not permitted or invalid URL format'), 'error') + else: + logger.error(f"add_watch: URL '{url}' is not permitted or invalid, skipping.") return None # Check PAGE_WATCH_LIMIT if set diff --git a/changedetectionio/tests/conftest.py b/changedetectionio/tests/conftest.py index 9212681c9..6da101c46 100644 --- a/changedetectionio/tests/conftest.py +++ b/changedetectionio/tests/conftest.py @@ -13,6 +13,10 @@ import sys # When test server is slow/unresponsive, workers fail fast instead of holding UUIDs for 45s # This prevents exponential priority growth from repeated deferrals (priority × 10 each defer) os.environ['DEFAULT_SETTINGS_REQUESTS_TIMEOUT'] = '5' +# Test server runs on localhost (127.0.0.1) which is a private IP. +# Allow it globally so all existing tests keep working; test_ssrf_protection +# uses monkeypatch to temporarily override this for its own assertions. +os.environ['ALLOW_IANA_RESTRICTED_ADDRESSES'] = 'true' from changedetectionio.flask_app import init_app_secret, changedetection_app from changedetectionio.tests.util import live_server_setup, new_live_server_setup diff --git a/changedetectionio/tests/test_security.py b/changedetectionio/tests/test_security.py index a76944cca..b7e5e2b5e 100644 --- a/changedetectionio/tests/test_security.py +++ b/changedetectionio/tests/test_security.py @@ -1,4 +1,5 @@ import os +import pytest from flask import url_for @@ -579,3 +580,131 @@ def test_static_directory_traversal(client, live_server, measure_memory_usage, d # Should get 403 (not authenticated) or 404 (file not found), not a path traversal assert res.status_code in [403, 404] + +def test_ssrf_private_ip_blocked(client, live_server, monkeypatch, measure_memory_usage, datastore_path): + """ + SSRF protection: IANA-reserved/private IP addresses must be blocked by default. + + Covers: + 1. is_private_hostname() correctly classifies all reserved ranges + 2. is_safe_valid_url() rejects private-IP URLs at add-time (env var off) + 3. is_safe_valid_url() allows private-IP URLs when ALLOW_IANA_RESTRICTED_ADDRESSES=true + 4. UI form rejects private-IP URLs and shows the standard error message + 5. Requests fetcher blocks fetch-time DNS rebinding (fresh check on every fetch) + 6. Requests fetcher blocks redirects that lead to a private IP (open-redirect bypass) + + conftest.py sets ALLOW_IANA_RESTRICTED_ADDRESSES=true globally so the test + server (localhost) keeps working for all other tests. monkeypatch temporarily + overrides it to 'false' here, and is automatically restored after the test. + """ + from unittest.mock import patch, MagicMock + from changedetectionio.validate_url import is_safe_valid_url, is_private_hostname + + monkeypatch.setenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false') + # Clear any URL results cached while the env var was 'true' + is_safe_valid_url.cache_clear() + + # ------------------------------------------------------------------ + # 1. is_private_hostname() — unit tests across all reserved ranges + # ------------------------------------------------------------------ + private_hosts = [ + '127.0.0.1', # loopback + '10.0.0.1', # RFC 1918 + '172.16.0.1', # RFC 1918 + '192.168.1.1', # RFC 1918 + '169.254.169.254', # link-local / AWS metadata endpoint + '::1', # IPv6 loopback + 'fc00::1', # IPv6 unique local + 'fe80::1', # IPv6 link-local + ] + for host in private_hosts: + assert is_private_hostname(host), f"{host} should be identified as private/reserved" + + for host in ['8.8.8.8', '1.1.1.1']: + assert not is_private_hostname(host), f"{host} should be identified as public" + + # ------------------------------------------------------------------ + # 2. is_safe_valid_url() blocks private-IP URLs (env var off) + # ------------------------------------------------------------------ + blocked_urls = [ + 'http://127.0.0.1/', + 'http://10.0.0.1/', + 'http://172.16.0.1/', + 'http://192.168.1.1/', + 'http://169.254.169.254/', + 'http://169.254.169.254/latest/meta-data/iam/security-credentials/', + 'http://[::1]/', + 'http://[fc00::1]/', + 'http://[fe80::1]/', + ] + for url in blocked_urls: + assert not is_safe_valid_url(url), f"{url} should be blocked by is_safe_valid_url" + + # ------------------------------------------------------------------ + # 3. ALLOW_IANA_RESTRICTED_ADDRESSES=true bypasses the block + # ------------------------------------------------------------------ + monkeypatch.setenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'true') + is_safe_valid_url.cache_clear() + assert is_safe_valid_url('http://127.0.0.1/'), \ + "Private IP should be allowed when ALLOW_IANA_RESTRICTED_ADDRESSES=true" + + # Restore the block for the remaining assertions + monkeypatch.setenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false') + is_safe_valid_url.cache_clear() + + # ------------------------------------------------------------------ + # 4. UI form rejects private-IP URLs + # ------------------------------------------------------------------ + for url in ['http://127.0.0.1/', 'http://169.254.169.254/latest/meta-data/']: + res = client.post( + url_for('ui.ui_views.form_quick_watch_add'), + data={'url': url, 'tags': ''}, + follow_redirects=True + ) + assert b'Watch protocol is not permitted or invalid URL format' in res.data, \ + f"UI should reject {url}" + + # ------------------------------------------------------------------ + # 5. Fetch-time DNS-rebinding check in the requests fetcher + # Simulates: URL passed add-time validation with a public IP, but + # by fetch time DNS has been rebound to a private IP. + # ------------------------------------------------------------------ + from changedetectionio.content_fetchers.requests import fetcher as RequestsFetcher + + f = RequestsFetcher() + + with patch('changedetectionio.content_fetchers.requests.is_private_hostname', return_value=True): + with pytest.raises(Exception, match='private/reserved'): + f._run_sync( + url='http://example.com/', + timeout=5, + request_headers={}, + request_body=None, + request_method='GET', + ) + + # ------------------------------------------------------------------ + # 6. Redirect-to-private-IP blocked (open-redirect SSRF bypass) + # Public host returns a 302 pointing at an IANA-reserved address. + # ------------------------------------------------------------------ + mock_redirect = MagicMock() + mock_redirect.is_redirect = True + mock_redirect.status_code = 302 + mock_redirect.headers = {'Location': 'http://169.254.169.254/latest/meta-data/'} + + def _private_only_for_redirect(hostname): + # Initial host is "public"; the redirect target is private + return hostname in {'169.254.169.254', '10.0.0.1', '172.16.0.1', + '192.168.0.1', '127.0.0.1', '::1'} + + with patch('changedetectionio.content_fetchers.requests.is_private_hostname', + side_effect=_private_only_for_redirect): + with patch('requests.Session.request', return_value=mock_redirect): + with pytest.raises(Exception, match='Redirect blocked'): + f._run_sync( + url='http://example.com/', + timeout=5, + request_headers={}, + request_body=None, + request_method='GET', + ) diff --git a/changedetectionio/validate_url.py b/changedetectionio/validate_url.py index 1cb208d2c..e942b1615 100644 --- a/changedetectionio/validate_url.py +++ b/changedetectionio/validate_url.py @@ -1,3 +1,5 @@ +import ipaddress +import socket from functools import lru_cache from loguru import logger from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode @@ -56,6 +58,23 @@ def normalize_url_encoding(url): return url +def is_private_hostname(hostname): + """Return True if hostname resolves to an IANA-restricted (private/reserved) IP address. + + Fails closed: unresolvable hostnames return True (block them). + Never cached — callers that need fresh DNS resolution (e.g. at fetch time) can call + this directly without going through the lru_cached is_safe_valid_url(). + """ + try: + for info in socket.getaddrinfo(hostname, None): + ip = ipaddress.ip_address(info[4][0]) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + return True + except socket.gaierror: + return True + return False + + @lru_cache(maxsize=10000) def is_safe_valid_url(test_url): from changedetectionio import strtobool @@ -119,4 +138,12 @@ def is_safe_valid_url(test_url): logger.warning(f'URL f"{test_url}" failed validation, aborting.') return False + # Block IANA-restricted (private/reserved) IP addresses unless explicitly allowed. + # This is an add-time check; fetch-time re-validation in requests.py handles DNS rebinding. + if not strtobool(os.getenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false')): + parsed = urlparse(test_url) + if parsed.hostname and is_private_hostname(parsed.hostname): + logger.warning(f'URL "{test_url}" resolves to a private/reserved IP address, aborting.') + return False + return True