import asyncio import re import hashlib from changedetectionio.browser_steps.browser_steps import browser_steps_get_valid_steps from changedetectionio.content_fetchers.base import Fetcher from changedetectionio.strtobool import strtobool from changedetectionio.validate_url import is_private_hostname, is_url_private_or_parser_confused from copy import deepcopy from abc import abstractmethod import os from urllib.parse import urlparse from loguru import logger SCREENSHOT_FORMAT_JPEG = 'JPEG' SCREENSHOT_FORMAT_PNG = 'PNG' class difference_detection_processor(): browser_steps = None datastore = None fetcher = None screenshot = None watch = None xpath_data = None preferred_proxy = None screenshot_format = SCREENSHOT_FORMAT_JPEG last_raw_content_checksum = None def __init__(self, datastore, watch_uuid): self.datastore = datastore self.watch_uuid = watch_uuid # Create a stable snapshot of the watch for processing # Why deepcopy? # 1. Prevents "dict changed during iteration" errors if watch is modified during processing # 2. Preserves Watch object with properties (.link, .is_pdf, etc.) - can't use dict() # 3. Safe now: Watch.__deepcopy__() shares datastore ref (no memory leak) but copies dict data self.watch = deepcopy(self.datastore.data['watching'].get(watch_uuid)) # Generic fetcher that should be extended (requests, playwright etc) self.fetcher = Fetcher() # Load the last raw content checksum from file self.read_last_raw_content_checksum() def update_last_raw_content_checksum(self, checksum): """ Save the raw content MD5 checksum to file. This is used for skip logic - avoid reprocessing if raw HTML unchanged. """ if not checksum: return watch = self.datastore.data['watching'].get(self.watch_uuid) if not watch: return data_dir = watch.data_dir if not data_dir: return watch.ensure_data_dir_exists() checksum_file = os.path.join(data_dir, 'last-checksum.txt') try: with open(checksum_file, 'w', encoding='utf-8') as f: f.write(checksum) self.last_raw_content_checksum = checksum except IOError as e: logger.warning(f"Failed to write checksum file for {self.watch_uuid}: {e}") def read_last_raw_content_checksum(self): """ Read the last raw content MD5 checksum from file. Returns None if file doesn't exist (first run) or can't be read. """ watch = self.datastore.data['watching'].get(self.watch_uuid) if not watch: self.last_raw_content_checksum = None return data_dir = watch.data_dir if not data_dir: self.last_raw_content_checksum = None return checksum_file = os.path.join(data_dir, 'last-checksum.txt') if not os.path.isfile(checksum_file): self.last_raw_content_checksum = None return try: with open(checksum_file, 'r', encoding='utf-8') as f: self.last_raw_content_checksum = f.read().strip() except IOError as e: logger.warning(f"Failed to read checksum file for {self.watch_uuid}: {e}") self.last_raw_content_checksum = None async def validate_iana_url(self): """Pre-flight SSRF check — runs DNS lookup in executor to avoid blocking the event loop. Covers all fetchers (requests, playwright, puppeteer, plugins) since every fetch goes through call_browser(). """ if strtobool(os.getenv('ALLOW_IANA_RESTRICTED_ADDRESSES', 'false')): return loop = asyncio.get_running_loop() # Use the parser-agnostic check so urlparse/urllib3 differentials (GHSA-rph4-96w6-q594) # can't slip a private/internal hostname past this pre-flight gate. if await loop.run_in_executor(None, is_url_private_or_parser_confused, self.watch.link): raise Exception( f"Fetch blocked: '{self.watch.link}' resolves to a private/reserved IP address " f"or contains a parser-differential payload. " f"Set ALLOW_IANA_RESTRICTED_ADDRESSES=true to allow." ) def _consume_preloaded_fetch(self): """One-shot: if the Add Watch page parked a freshly-fetched snapshot for this watch (html + screenshot + xpath, see add_watch_ui/snapshot), populate self.fetcher from it instead of hitting the network. The marker file is deleted after use so every subsequent check fetches live. Returns True if a preload was consumed (caller should skip the network fetch). """ import json, zlib data_dir = self.watch.data_dir if not data_dir: return False preload_path = os.path.join(data_dir, 'preload-fetch.json') if not os.path.isfile(preload_path): return False try: with open(preload_path, 'r', encoding='utf-8') as f: meta = json.load(f) except Exception as e: logger.warning(f"Could not read preloaded fetch for {self.watch.get('uuid')}: {e}") try: os.unlink(preload_path) except OSError: pass return False # Always delete first - this is one-shot regardless of what happens next. try: os.unlink(preload_path) except OSError: pass content = meta.get('content') self.fetcher.content = content self.fetcher.raw_content = content.encode('utf-8', errors='replace') if isinstance(content, str) else content self.fetcher.status_code = meta.get('status_code', 200) self.fetcher.headers = meta.get('headers') or {'content-type': 'text/html'} # Screenshot + xpath were migrated alongside in final on-disk format - reuse them. screenshot_path = os.path.join(data_dir, 'last-screenshot.png') if os.path.isfile(screenshot_path): with open(screenshot_path, 'rb') as f: self.fetcher.screenshot = f.read() elements_path = os.path.join(data_dir, 'elements.deflate') if os.path.isfile(elements_path): try: with open(elements_path, 'rb') as f: self.fetcher.xpath_data = json.loads(zlib.decompress(f.read())) except Exception as e: logger.warning(f"Could not load preloaded xpath data for {self.watch.get('uuid')}: {e}") logger.info(f"Using preloaded Add-Watch snapshot for {self.watch.get('uuid')} - skipping network fetch") return True async def call_browser(self, preferred_proxy_id=None): from requests.structures import CaseInsensitiveDict url = self.watch.link # Protect against file:, file:/, file:// access, check the real "link" without any meta "source:" etc prepended. if re.search(r'^file:', url.strip(), re.IGNORECASE): if not strtobool(os.getenv('ALLOW_FILE_URI', 'false')): raise Exception( "file:// type access is denied for security reasons." ) await self.validate_iana_url() # Proxy ID "key" preferred_proxy_id = preferred_proxy_id if preferred_proxy_id else self.datastore.get_preferred_proxy_for_watch( uuid=self.watch.get('uuid')) # Resolve which content fetcher this watch should use. This is the single source of # truth (watch -> group -> system, extra_browser_/pdf/browser_steps overrides etc); # the resolved backend name is stamped onto the fetcher instance below. from changedetectionio.content_fetchers import resolve_content_fetcher fetcher_obj, prefer_fetch_backend, custom_browser_connection_url = resolve_content_fetcher( watch=self.watch, datastore=self.datastore) proxy_url = None if preferred_proxy_id: # Custom browser endpoints should NOT have a proxy added if not prefer_fetch_backend.startswith('extra_browser_'): proxy_url = self.datastore.proxy_list.get(preferred_proxy_id).get('url') logger.debug(f"Selected proxy key '{preferred_proxy_id}' as proxy URL '{proxy_url}' for {url}") else: logger.debug("Skipping adding proxy data when custom Browser endpoint is specified. ") logger.debug(f"Using proxy '{proxy_url}' for {self.watch['uuid']}") # Now call the fetcher (playwright/requests/etc) with arguments that only a fetcher would need. # When browser_connection_url is None, it method should default to working out whats the best defaults (os env vars etc) self.fetcher = fetcher_obj(proxy_override=proxy_url, custom_browser_connection_url=custom_browser_connection_url, screenshot_format=self.screenshot_format ) # Stamp the resolved backend name so downstream consumers (processors, plugins) # can read it directly instead of re-deriving it from the fetcher class name. self.fetcher.backend_name = prefer_fetch_backend if self.watch.has_browser_steps: self.fetcher.browser_steps = browser_steps_get_valid_steps(self.watch.get('browser_steps', [])) self.fetcher.browser_steps_screenshot_path = os.path.join(self.datastore.datastore_path, self.watch.get('uuid')) # Tweak the base config with the per-watch ones from changedetectionio.jinja2_custom import render as jinja_render request_headers = CaseInsensitiveDict() ua = self.datastore.data['settings']['requests'].get('default_ua') if ua and ua.get(prefer_fetch_backend): request_headers.update({'User-Agent': ua.get(prefer_fetch_backend)}) request_headers.update(self.watch.get('headers', {})) request_headers.update(self.datastore.get_all_base_headers()) request_headers.update(self.datastore.get_all_headers_in_textfile_for_watch(uuid=self.watch.get('uuid'))) # https://github.com/psf/requests/issues/4525 # Requests doesnt yet support brotli encoding, so don't put 'br' here, be totally sure that the user cannot # do this by accident. if 'Accept-Encoding' in request_headers and "br" in request_headers['Accept-Encoding']: request_headers['Accept-Encoding'] = request_headers['Accept-Encoding'].replace(', br', '') for header_name in request_headers: request_headers.update({header_name: jinja_render(template_str=request_headers.get(header_name))}) timeout = self.datastore.data['settings']['requests'].get('timeout') request_body = self.watch.get('body') if request_body: request_body = jinja_render(template_str=self.watch.get('body')) request_method = self.watch.get('method') ignore_status_codes = self.watch.get('ignore_status_codes', False) # Configurable per-watch or global extra delay before extracting text (for webDriver types) system_webdriver_delay = self.datastore.data['settings']['application'].get('webdriver_delay', None) if self.watch.get('webdriver_delay'): self.fetcher.render_extract_delay = self.watch.get('webdriver_delay') elif system_webdriver_delay is not None: self.fetcher.render_extract_delay = system_webdriver_delay if self.watch.get('webdriver_js_execute_code') is not None and self.watch.get('webdriver_js_execute_code').strip(): self.fetcher.webdriver_js_execute_code = self.watch.get('webdriver_js_execute_code') # Requests for PDF's, images etc should be passwd the is_binary flag is_binary = self.watch.is_pdf # And here we go! call the right browser with browser-specific settings empty_pages_are_a_change = self.datastore.data['settings']['application'].get('empty_pages_are_a_change', False) # All fetchers are now async await self.fetcher.run( current_include_filters=self.watch.get('include_filters'), empty_pages_are_a_change=empty_pages_are_a_change, fetch_favicon=self.watch.favicon_is_expired(), ignore_status_codes=ignore_status_codes, is_binary=is_binary, request_body=request_body, request_headers=request_headers, request_method=request_method, screenshot_format=self.screenshot_format, timeout=timeout, url=url, watch_uuid=self.watch_uuid, ) # @todo .quit here could go on close object, so we can run JS if change-detected await self.fetcher.quit(watch=self.watch) # Sanitize lone surrogates - these can appear when servers return malformed/mixed-encoding # content that gets decoded into surrogate characters (e.g. \udcad). Without this, # encode('utf-8') raises UnicodeEncodeError downstream in checksums, diffs, file writes, etc. # Covers all fetchers (requests, playwright, puppeteer, selenium) in one place. # Also note: By this point we SHOULD know the original encoding so it can safely convert to utf-8 for the rest of the app. # See: https://github.com/dgtlmoon/changedetection.io/issues/3952 if self.fetcher.content and isinstance(self.fetcher.content, str): self.fetcher.content = self.fetcher.content.encode('utf-8', errors='replace').decode('utf-8') # After init, call run_changedetection() which will do the actual change-detection @staticmethod def _resolve_watch_config_path(data_dir, filename): """Resolve `filename` inside `data_dir`, refusing anything that escapes it. Security: callers derive `filename` from watch['processor'] (see processors/save_processor_config), and that value is not enum-validated on every write path - so it must be treated as untrusted here. os.path.join() will happily accept '../../../../tmp/pwned', which previously escaped the watch directory and allowed an arbitrary-path JSON file write (and read) as the app user. Returns the absolute path, or None if it is not safely contained. """ import os if not filename or filename in ('.', '..'): logger.error(f"Refusing unsafe watch config filename {filename!r}") return None # Must be a bare filename - no directory component, no separator of either flavour if filename != os.path.basename(filename) or '/' in filename or '\\' in filename: logger.error(f"Refusing watch config filename with a path component: {filename!r}") return None # realpath both sides so a symlink planted inside data_dir cannot redirect the write base = os.path.realpath(data_dir) filepath = os.path.realpath(os.path.join(base, filename)) if os.path.dirname(filepath) != base: logger.error(f"Refusing watch config path outside the watch directory: {filepath!r}") return None return filepath def get_extra_watch_config(self, filename): """ Read processor-specific JSON config file from watch data directory. Args: filename: Name of JSON file (e.g., "visual_ssim_score.json") Returns: dict: Parsed JSON data, or empty dict if file doesn't exist """ import json import os watch = self.datastore.data['watching'].get(self.watch_uuid) data_dir = watch.data_dir if not data_dir: return {} filepath = self._resolve_watch_config_path(data_dir, filename) if not filepath: return {} if not os.path.isfile(filepath): return {} try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, IOError) as e: logger.warning(f"Failed to read extra watch config {filename}: {e}") return {} def update_extra_watch_config(self, filename, data, merge=True): """ Write processor-specific JSON config file to watch data directory. Args: filename: Name of JSON file (e.g., "visual_ssim_score.json") data: Dictionary to serialize as JSON merge: If True, merge with existing data; if False, overwrite completely """ import json import os watch = self.datastore.data['watching'].get(self.watch_uuid) data_dir = watch.data_dir if not data_dir: logger.warning(f"Cannot save extra watch config {filename}: no data_dir") return # Ensure directory exists watch.ensure_data_dir_exists() filepath = self._resolve_watch_config_path(data_dir, filename) if not filepath: return try: # If merge is enabled, read existing data first existing_data = {} if merge and os.path.isfile(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: existing_data = json.load(f) except (json.JSONDecodeError, IOError) as e: logger.warning(f"Failed to read existing config for merge: {e}") # Merge new data with existing if merge: existing_data.update(data) data_to_save = existing_data else: data_to_save = data # Write the data with open(filepath, 'w', encoding='utf-8') as f: json.dump(data_to_save, f, indent=2) except IOError as e: logger.error(f"Failed to write extra watch config {filename}: {e}") def get_raw_document_checksum(self): checksum = None if self.fetcher.content: checksum = hashlib.md5(self.fetcher.content.encode('utf-8')).hexdigest() return checksum @abstractmethod def run_changedetection(self, watch, force_reprocess=False): update_obj = {'last_notification_error': False, 'last_error': False} some_data = 'xxxxx' update_obj["previous_md5"] = hashlib.md5(some_data.encode('utf-8')).hexdigest() changed_detected = False return changed_detected, update_obj, ''.encode('utf-8')