Security: Fix SSRF vulnerabilities in URL fetching

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:

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:

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))

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(