Arthur031221 · Oct 15, 2026
diff --git a/scrapegraphai/docloaders/chromium.py b/scrapegraphai/docloaders/chromium.py
@@ -6,6 +6,7 @@
 from langchain_core.documents import Document
 
 from ..utils import Proxy, dynamic_import, get_logger, parse_or_search_proxy
+from ..utils.url_safety import validate_url
 
 logger = get_logger("web-loader")
 
@@ -36,6 +37,16 @@ def _warn_on_error_status(response: Any, url: str) -> None:
         )
 
 
+async def _block_unsafe_request(route: Any) -> None:
+    """Abort browser requests whose host resolves to a metadata address."""
+    try:
+        await asyncio.to_thread(validate_url, route.request.url)
+    except ValueError:
+        await route.abort()
+    else:
+        await route.continue_()
+
+
 class ChromiumLoader:
     """Scrapes HTML pages from URLs using a (headless) instance of the
     Chromium web driver with proxy protection.
@@ -276,8 +287,10 @@ async def ascrape_playwright_scroll(
                         raise ValueError(f"Invalid browser name: {browser_name}")
                     context = await browser.new_context()
                     await Malenia.apply_stealth(context)
+                    await context.route("**/*", _block_unsafe_request)
                     page = await context.new_page()
                     response = await page.goto(url, wait_until="domcontentloaded")
+                    await asyncio.to_thread(validate_url, page.url)
                     _warn_on_error_status(response, url)
                     await page.wait_for_load_state(self.load_state)
 
@@ -332,6 +345,7 @@ async def ascrape_playwright_scroll(
                                 break
 
                     results = await page.content()
+                    await asyncio.to_thread(validate_url, page.url)
                     break
 
             except (aiohttp.ClientError, asyncio.TimeoutError, Exception) as e:
@@ -390,11 +404,14 @@ async def ascrape_playwright(self, url: str, browser_name: str = "chromium") ->
                         ignore_https_errors=True,
                     )
                     await Malenia.apply_stealth(context)
+                    await context.route("**/*", _block_unsafe_request)
                     page = await context.new_page()
                     response = await page.goto(url, wait_until="domcontentloaded")
+                    await asyncio.to_thread(validate_url, page.url)
                     _warn_on_error_status(response, url)
                     await page.wait_for_load_state(self.load_state)
                     results = await page.content()
+                    await asyncio.to_thread(validate_url, page.url)
                     logger.info("Content scraped")
                     await browser.close()
                     return results
@@ -448,10 +465,13 @@ async def ascrape_with_js_support(
                     context = await browser.new_context(
                         storage_state=self.storage_state
                     )
+                    await context.route("**/*", _block_unsafe_request)
                     page = await context.new_page()
                     response = await page.goto(url, wait_until="networkidle")
+                    await asyncio.to_thread(validate_url, page.url)
                     _warn_on_error_status(response, url)
                     results = await page.content()
+                    await asyncio.to_thread(validate_url, page.url)
                     logger.info("Content scraped after JavaScript rendering")
                     return results
             except (aiohttp.ClientError, asyncio.TimeoutError, Exception) as e:

diff --git a/scrapegraphai/nodes/fetch_node.py b/scrapegraphai/nodes/fetch_node.py
@@ -13,6 +13,7 @@
 from ..docloaders import ChromiumLoader
 from ..utils.cleanup_html import cleanup_html
 from ..utils.convert_to_md import convert_to_md
+from ..utils.url_safety import validate_redirect, validate_url
 from .base_node import BaseNode
 
 
@@ -276,17 +277,25 @@ def handle_web_source(self, state, source):
         dict: The updated state with the processed content.
 
         Raises:
-        ValueError: If the fetched HTML content is empty or contains only whitespace.
+        ValueError: If the fetched HTML content is empty or contains only whitespace,
+        or if the URL points to a link-local or cloud metadata address.
         """
 
+        if not (self.browser_base or self.scrape_do):
+            validate_url(source)
+
         self.logger.info(f"--- (Fetching HTML from: {source}) ---")
         if self.use_soup:
             # Apply configured timeout to blocking HTTP requests. If timeout is None,
             # don't pass the timeout argument (requests will block until completion).
             if self.timeout is None:
-                response = requests.get(source)
+                response = requests.get(source, hooks={"response": validate_redirect})
             else:
-                response = requests.get(source, timeout=self.timeout)
+                response = requests.get(
+                    source,
+                    timeout=self.timeout,
+                    hooks={"response": validate_redirect},
+                )
             if response.status_code == 200:
                 if not response.text.strip():
                     raise ValueError("No HTML body content found in the response.")

diff --git a/scrapegraphai/utils/url_safety.py b/scrapegraphai/utils/url_safety.py
@@ -0,0 +1,61 @@
+"""
+Module for refusing scrape targets that point at cloud metadata endpoints
+"""
+
+import ipaddress
+import socket
+from urllib.parse import urljoin, urlparse
+
+# Metadata services that sit outside the link-local ranges
+_METADATA_ADDRESSES = {
+    ipaddress.ip_address("fd00:ec2::254"),
+    ipaddress.ip_address("100.100.100.200"),
+}
+
+
+def validate_url(url: str) -> None:
+    """
+    Raises ValueError if the host of the url resolves to a link-local address
+    (169.254.0.0/16, fe80::/10) or to a known cloud metadata address.
+    Hosts that do not resolve are left to fail when they are fetched.
+
+    Args:
+        url (str): The url about to be fetched.
+
+    Raises:
+        ValueError: If any address the host resolves to is not allowed.
+    """
+    hostname = urlparse(url).hostname
+    if not hostname:
+        return
+
+    try:
+        # urllib3 drops an IPv6 zone id (fe80::1%25eth0) before connecting
+        infos = socket.getaddrinfo(hostname.split("%")[0], None)
+    except socket.gaierror:
+        return
+
+    for info in infos:
+        address = ipaddress.ip_address(info[4][0].split("%")[0])
+        if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
+            address = address.ipv4_mapped
+        if address.is_link_local or address in _METADATA_ADDRESSES:
+            raise ValueError(
+                "The url points to a link-local or cloud metadata address "
+                "and will not be fetched."
+            )
+
+
+def validate_redirect(response, **kwargs) -> None:
+    """
+    Response hook for requests that checks the target of a redirect before
+    requests follows it.
+
+    Args:
+        response: The response that was just received.
+
+    Raises:
+        ValueError: If the redirect target is not allowed.
+    """
+    if response.is_redirect:
+        validate_url(urljoin(response.url, response.headers["location"]))