diff --git a/extensions/superbooga/download_urls.py b/extensions/superbooga/download_urls.py index b28fea42..280fc2fa 100644 --- a/extensions/superbooga/download_urls.py +++ b/extensions/superbooga/download_urls.py @@ -1,16 +1,13 @@ import concurrent.futures -import requests - -from modules.web_search import _validate_url +from modules.web_search import safe_get def download_single(url): - _validate_url(url) headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3' } - response = requests.get(url, headers=headers, timeout=5) + response = safe_get(url, headers=headers, timeout=5) if response.status_code == 200: return response.content else: diff --git a/extensions/superboogav2/download_urls.py b/extensions/superboogav2/download_urls.py index 4d8b98b1..faed8611 100644 --- a/extensions/superboogav2/download_urls.py +++ b/extensions/superboogav2/download_urls.py @@ -1,19 +1,17 @@ import concurrent.futures import re -import requests from bs4 import BeautifulSoup import extensions.superboogav2.parameters as parameters -from modules.web_search import _validate_url +from modules.web_search import safe_get from .data_processor import process_and_add_to_collector from .utils import create_metadata_source def _download_single(url): - _validate_url(url) - response = requests.get(url, timeout=5) + response = safe_get(url, timeout=5) if response.status_code == 200: return response.content else: diff --git a/modules/image_utils.py b/modules/image_utils.py index b3138790..51f09f7d 100644 --- a/modules/image_utils.py +++ b/modules/image_utils.py @@ -76,19 +76,8 @@ def process_message_content(content: Any) -> Tuple[str, List[Image.Image]]: elif image_url.startswith('http'): # Support external URLs try: - import requests - from urllib.parse import urljoin - from modules.web_search import _validate_url - _validate_url(image_url) - url = image_url - for _ in range(5): - response = requests.get(url, timeout=10, allow_redirects=False) - if response.is_redirect and 'Location' in response.headers: - url = urljoin(url, response.headers['Location']) - _validate_url(url) - else: - break - + from modules.web_search import safe_get + response = safe_get(image_url, timeout=10) response.raise_for_status() image_data = response.content image = Image.open(io.BytesIO(image_data)) diff --git a/modules/web_search.py b/modules/web_search.py index 2902c7c0..dd5d2be4 100644 --- a/modules/web_search.py +++ b/modules/web_search.py @@ -16,10 +16,18 @@ from modules.logging_colors import logger def _validate_url(url): """Validate that a URL is safe to fetch (not targeting private/internal networks).""" + # Reject characters that cause parsing discrepancies between urlparse and requests, + # which can be exploited to bypass SSRF protections (GHSA-27xf-58m5-vxmc). + if '\\' in url: + raise ValueError("Invalid URL: backslashes are not allowed") + parsed = urlparse(url) if parsed.scheme not in ('http', 'https'): raise ValueError(f"Unsupported URL scheme: {parsed.scheme}") + if '@' in parsed.netloc: + raise ValueError("Invalid URL: userinfo (credentials) in URLs is not allowed") + hostname = parsed.hostname if not hostname: raise ValueError("No hostname in URL") @@ -34,6 +42,20 @@ def _validate_url(url): raise ValueError(f"Could not resolve hostname: {hostname}") +def safe_get(url, headers=None, timeout=10, max_redirects=5): + """Fetch a URL with SSRF-safe redirect handling. Validates every hop.""" + _validate_url(url) + for _ in range(max_redirects): + response = requests.get(url, headers=headers, timeout=timeout, allow_redirects=False) + if response.is_redirect and 'Location' in response.headers: + url = urljoin(url, response.headers['Location']) + _validate_url(url) + else: + return response + + raise ValueError(f"Too many redirects (max {max_redirects})") + + def get_current_timestamp(): """Returns the current time in 24-hour format""" return datetime.now().strftime('%b %d, %Y %H:%M') @@ -46,19 +68,10 @@ def download_web_page(url, timeout=10, include_links=False): import trafilatura try: - _validate_url(url) headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36' } - max_redirects = 5 - for _ in range(max_redirects): - response = requests.get(url, headers=headers, timeout=timeout, allow_redirects=False) - if response.is_redirect and 'Location' in response.headers: - url = urljoin(url, response.headers['Location']) - _validate_url(url) - else: - break - + response = safe_get(url, headers=headers, timeout=timeout) response.raise_for_status() result = trafilatura.extract(