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.

This commit is contained in:
dgtlmoon
2026-02-23 21:56:43 +01:00
parent a385c89abf
commit fe7aa38c65
6 changed files with 204 additions and 5 deletions
+37 -1
View File
@@ -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:
+2 -2
View File
@@ -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
+5 -2
View File
@@ -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
+4
View File
@@ -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
+129
View File
@@ -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',
)
+27
View File
@@ -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