import ipaddress import socket from functools import lru_cache from loguru import logger from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode def normalize_url_encoding(url): """ Safely encode a URL's query parameters, regardless of whether they're already encoded. Why this is necessary: URLs can arrive in various states - some with already encoded query parameters (%20 for spaces), some with unencoded parameters (literal spaces), or a mix of both. The validators.url() function requires proper encoding, but simply encoding an already-encoded URL would double-encode it (e.g., %20 would become %2520). This function solves the problem by: 1. Parsing the URL to extract query parameters 2. parse_qsl() automatically decodes parameters if they're encoded 3. urlencode() re-encodes them properly 4. Returns a consistently encoded URL that will pass validation Example: - Input: "http://example.com/test?time=2025-10-28 09:19" (space not encoded) - Output: "http://example.com/test?time=2025-10-28+09%3A19" (properly encoded) - Input: "http://example.com/test?time=2025-10-28%2009:19" (already encoded) - Output: "http://example.com/test?time=2025-10-28+09%3A19" (properly encoded) Returns a properly encoded URL string. """ try: # Parse the URL into components (scheme, netloc, path, params, query, fragment) parsed = urlparse(url) # Parse query string - this automatically decodes it if encoded # parse_qsl handles both encoded and unencoded query strings gracefully query_params = parse_qsl(parsed.query, keep_blank_values=True) # Re-encode the query string properly using standard URL encoding encoded_query = urlencode(query_params, safe='') # Reconstruct the URL with properly encoded query string normalized = urlunparse(( parsed.scheme, parsed.netloc, parsed.path, parsed.params, encoded_query, # Use the re-encoded query parsed.fragment )) return normalized except Exception as e: # If parsing fails for any reason, return original URL logger.debug(f"URL normalization failed for '{url}': {e}") return url def is_private_hostname(hostname): """Return True if hostname resolves to an IANA-restricted (private/reserved) IP address. Unresolvable hostnames return False (allow them) — DNS may be temporarily unavailable or the domain not yet live. The actual DNS rebinding attack is mitigated by fetch-time re-validation in requests.py, not by blocking unresolvable domains at add-time. 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: logger.warning(f"Hostname '{hostname} - {ip} - ip.is_private = {ip.is_private}, ip.is_loopback = {ip.is_loopback}, ip.is_link_local = {ip.is_link_local}, ip.is_reserved = {ip.is_reserved}") return True except socket.gaierror as e: logger.warning(f"{hostname} error checking {str(e)}") return False logger.info(f"Hostname '{hostname}' is NOT private/IANA restricted.") return False def extract_url_hostnames(url): """Return every hostname this URL could resolve to under different URL parsers. Why: urllib's urlparse() and urllib3's parse_url() disagree on URLs containing a backslash (e.g. http://INTERNAL:8888\\@PUBLIC/ — urlparse extracts PUBLIC, but urllib3/requests will actually connect to INTERNAL). Any SSRF check that trusts only one parser can be bypassed by the other. Callers should reject the fetch if ANY hostname returned here is private/reserved. See GHSA-rph4-96w6-q594. """ hostnames = set() try: h = urlparse(url).hostname if h: hostnames.add(h) except Exception: pass try: from urllib3.util.url import parse_url as _u3_parse_url u3 = _u3_parse_url(url) if u3.host: # urllib3 keeps IPv6 brackets in `.host`; strip them so socket.getaddrinfo() accepts the literal. hostnames.add(u3.host.strip('[]')) except Exception: pass return hostnames def is_url_private_or_parser_confused(url): """SSRF gate that defends against urlparse/urllib3 parser-differential attacks. Returns True (block the fetch) when: * the URL contains a backslash — no legitimate URL needs one, and it is the established vector for the parser-differential bypass (GHSA-rph4-96w6-q594), OR * any hostname produced by urlparse OR urllib3 resolves to a private/reserved IP. """ if '\\' in url: logger.warning(f"URL '{url}' contains a backslash — rejected to prevent urlparse/urllib3 parser-differential SSRF.") return True for hostname in extract_url_hostnames(url): if is_private_hostname(hostname): return True return False def is_llm_api_base_safe(api_base): """SSRF guard for the LLM `api_base` setting (GHSA-jrxm-qjfh-g54f). Returns (ok: bool, reason: str). Empty/None api_base is allowed (cloud providers don't need it). When ALLOW_IANA_RESTRICTED_ADDRESSES=true the check is bypassed so operators can intentionally point at local Ollama / vLLM / LM Studio. Call this from EVERY write path that accepts `llm.api_base` from the user — form validation, AJAX endpoints, and any future REST/import endpoint. The existing call sites are forms.py (validateLLMApiBaseSafe) and blueprint/settings/llm.py (both /models and /test). """ import os from changedetectionio.strtobool import strtobool from flask_babel import gettext if not api_base or not api_base.strip(): return True, '' if strtobool(os.getenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false')): return True, '' api_base = api_base.strip() if not is_safe_valid_url(api_base): return False, gettext("API Base URL is not a valid http(s) URL.") hostname = urlparse(api_base).hostname if hostname and is_private_hostname(hostname): return False, gettext( "API Base URL resolves to a private, loopback, link-local or reserved " "IP address and was blocked to prevent SSRF. To allow LLM endpoints on private networks " "(e.g. a local Ollama server) set the environment variable " "ALLOW_IANA_RESTRICTED_ADDRESSES=true and restart." ) return True, '' def is_safe_valid_url(test_url): from changedetectionio import strtobool from changedetectionio.jinja2_custom import render as jinja_render import os import re import validators # Validate input type first - must be a non-empty string if test_url is None: logger.warning('URL validation failed: URL is None') return False if not isinstance(test_url, str): logger.warning(f'URL validation failed: URL must be a string, got {type(test_url).__name__}') return False if not test_url.strip(): logger.warning('URL validation failed: URL is empty or whitespace only') return False # Per-request cache: same URL is often validated 2-3x per watchlist render (sort + display). # Flask's g is scoped to one request and auto-cleared on teardown, so dynamic Jinja2 URLs # like {{microtime()}} are always re-evaluated on the next request. # Falls back gracefully when called outside a request context (e.g. background workers). _cache_key = test_url try: from flask import g _cache = g.setdefault('_url_validation_cache', {}) if _cache_key in _cache: return _cache[_cache_key] except RuntimeError: _cache = None # No app context allow_file_access = strtobool(os.getenv('ALLOW_FILE_URI', 'false')) safe_protocol_regex = '^(http|https|ftp|file):' if allow_file_access else '^(http|https|ftp):' # See https://github.com/dgtlmoon/changedetection.io/issues/1358 # Remove 'source:' prefix so we dont get 'source:javascript:' etc # 'source:' is a valid way to tell us to return the source r = re.compile('^source:', re.IGNORECASE) test_url = r.sub('', test_url) # Check the actual rendered URL in case of any Jinja markup # Only run jinja_render when the URL actually contains Jinja2 syntax - creating a new # ImmutableSandboxedEnvironment is expensive and is called once per watch per page load if '{%' in test_url or '{{' in test_url: try: test_url = jinja_render(test_url) except Exception as e: logger.error(f'URL "{test_url}" is not correct Jinja2? {str(e)}') return False # Check query parameters and fragment if re.search(r'[<>]', test_url): logger.warning(f'URL "{test_url}" contains suspicious characters') return False # Reject backslashes — urllib's urlparse and urllib3's parse_url disagree on URLs containing # a backslash (e.g. http://INTERNAL:8888\@PUBLIC/), which is the documented SSRF bypass in # GHSA-rph4-96w6-q594. A backslash has no legitimate use in an HTTP URL, so block at add-time. if '\\' in test_url: logger.warning(f'URL "{test_url}" contains a backslash — rejected (parser-differential SSRF vector).') return False # Normalize URL encoding - handle both encoded and unencoded query parameters test_url = normalize_url_encoding(test_url) # Be sure the protocol is safe (no file, etcetc) pattern = re.compile(os.getenv('SAFE_PROTOCOL_REGEX', safe_protocol_regex), re.IGNORECASE) if not pattern.match(test_url.strip()): logger.warning(f'URL "{test_url}" is not safe, aborting.') return False # If hosts that only contain alphanumerics are allowed ("localhost" for example) allow_simplehost = not strtobool(os.getenv('BLOCK_SIMPLEHOSTS', 'False')) try: if not test_url.strip().lower().startswith('file:') and not validators.url(test_url, simple_host=allow_simplehost): logger.warning(f'URL "{test_url}" failed validation, aborting.') return False except validators.ValidationError: logger.warning(f'URL f"{test_url}" failed validation, aborting.') return False if _cache is not None: _cache[_cache_key] = True return True