fix(tools): pin SSRF checks to each redirect hop and peer IP (#6981)

* fix(tools): pin SSRF checks to each redirect hop and peer IP

validate_url only inspected the original URL string, so scraping fetches
could follow a 302 to an internal address or rebind DNS between check and
connect. Route safe_get through an HTTPAdapter that re-validates every hop
and connects to the authorised sockaddr, and let FORCE_SAFE_PATHS ignore a
tenant-supplied escape hatch on managed workers.

Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>

* Potential fix for pull request finding 'Except block handles 'BaseException''

Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

* test(azure): use a plain stand-in for Responses API delegate mocks

MagicMock instances are not reliably stored on Pydantic PrivateAttr via
BaseLLM.__setattr__, which left _responses_delegate as None and failed
last_response_id / reset_chain assertions on CI.

Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

docs/edge/ar/tools/web-scraping/scrapeelementfromwebsitetool.mdx

@@ -9,7 +9,7 @@ mode: "wide"
 
 ## الوصف
 
-أداة `ScrapeElementFromWebsiteTool` مصممة لاستخراج عناصر محددة من المواقع باستخدام محددات CSS. تسمح هذه الأداة لوكلاء CrewAI باستخراج محتوى مستهدف من صفحات الويب، مما يجعلها مفيدة لمهام استخراج البيانات حيث تكون أجزاء محددة فقط من صفحة الويب مطلوبة.
+أداة `ScrapeElementFromWebsiteTool` مصممة لاستخراج عناصر محددة من المواقع باستخدام محددات CSS. تسمح هذه الأداة لوكلاء CrewAI باستخراج محتوى مستهدف من صفحات الويب، مما يجعلها مفيدة لمهام استخراج البيانات حيث تكون أجزاء محددة فقط من صفحة الويب مطلوبة. تمر الطلبات عبر مساعد HTTP الآمن ضد SSRF في CrewAI: يتم فحص عنوان URL المطلوب وكل قفزة إعادة توجيه مقابل النطاقات الخاصة والمحجوزة (بما في ذلك بيانات تعريف السحابة)، ويُثبَّت اتصال TCP على عنوان IP الذي اجتاز هذا الفحص.
 
 ## التثبيت
 

docs/edge/ar/tools/web-scraping/scrapewebsitetool.mdx

@@ -16,6 +16,8 @@ mode: "wide"
 أداة مصممة لاستخراج وقراءة محتوى موقع محدد. قادرة على التعامل مع أنواع مختلفة من صفحات الويب عن طريق إجراء طلبات HTTP وتحليل محتوى HTML المستلم.
 يمكن أن تكون هذه الأداة مفيدة بشكل خاص لمهام استخراج البيانات من الويب وجمع البيانات أو استخراج معلومات محددة من المواقع.
 
+تمر الطلبات عبر مساعد HTTP الآمن ضد SSRF في CrewAI: يتم فحص عنوان URL المطلوب وكل قفزة إعادة توجيه مقابل النطاقات الخاصة والمحجوزة (بما في ذلك بيانات تعريف السحابة)، ويُثبَّت اتصال TCP على عنوان IP الذي اجتاز هذا الفحص.
+
 ## التثبيت
 
 ثبّت حزمة crewai_tools

docs/edge/en/tools/file-document/filereadtool.mdx

@@ -77,4 +77,4 @@ To let an agent read a directory tree outside the working directory, point `base
 file_read_tool = FileReadTool(base_dir='/data')
 ```
 
-As a last resort, setting `CREWAI_TOOLS_ALLOW_UNSAFE_PATHS=true` disables path validation. This applies process-wide to every crewai-tools tool, including the SSRF protections on URL-fetching tools, so prefer `base_dir`.
+As a last resort, setting `CREWAI_TOOLS_ALLOW_UNSAFE_PATHS=true` disables path validation. This applies process-wide to every crewai-tools tool, including the SSRF protections on URL-fetching tools, so prefer `base_dir`. Managed workers should set `CREWAI_TOOLS_FORCE_SAFE_PATHS=true` so a tenant cannot disable those checks by exporting the escape hatch.

docs/edge/en/tools/web-scraping/scrapeelementfromwebsitetool.mdx

@@ -9,7 +9,7 @@ mode: "wide"
 
 ## Description
 
-The `ScrapeElementFromWebsiteTool` is designed to extract specific elements from websites using CSS selectors. This tool allows CrewAI agents to scrape targeted content from web pages, making it useful for data extraction tasks where only specific parts of a webpage are needed.
+The `ScrapeElementFromWebsiteTool` is designed to extract specific elements from websites using CSS selectors. This tool allows CrewAI agents to scrape targeted content from web pages, making it useful for data extraction tasks where only specific parts of a webpage are needed. Fetches go through CrewAI's SSRF-safe HTTP helper: the requested URL and every redirect hop are checked against private and reserved ranges (including cloud metadata), and the TCP connection is pinned to an IP that passed that check.
 
 ## Installation
 

docs/edge/en/tools/web-scraping/scrapewebsitetool.mdx

@@ -16,6 +16,8 @@ mode: "wide"
 A tool designed to extract and read the content of a specified website. It is capable of handling various types of web pages by making HTTP requests and parsing the received HTML content. 
 This tool can be particularly useful for web scraping tasks, data collection, or extracting specific information from websites.
 
+Fetches go through CrewAI's SSRF-safe HTTP helper: the requested URL and every redirect hop are checked against private and reserved ranges (including cloud metadata), and the TCP connection is pinned to an IP that passed that check.
+
 ## Installation
 
 Install the crewai_tools package

docs/edge/ko/tools/web-scraping/scrapeelementfromwebsitetool.mdx

@@ -9,7 +9,7 @@ mode: "wide"
 
 ## 설명
 
-`ScrapeElementFromWebsiteTool`은 CSS 선택자를 사용하여 웹사이트에서 특정 요소를 추출하도록 설계되었습니다. 이 도구는 CrewAI 에이전트가 웹 페이지에서 타겟이 되는 콘텐츠를 스크래핑할 수 있게 하여, 웹페이지의 특정 부분만이 필요한 데이터 추출 작업에 유용합니다.
+`ScrapeElementFromWebsiteTool`은 CSS 선택자를 사용하여 웹사이트에서 특정 요소를 추출하도록 설계되었습니다. 이 도구는 CrewAI 에이전트가 웹 페이지에서 타겟이 되는 콘텐츠를 스크래핑할 수 있게 하여, 웹페이지의 특정 부분만이 필요한 데이터 추출 작업에 유용합니다. 가져오기는 CrewAI의 SSRF 안전 HTTP 헬퍼를 거칩니다. 요청된 URL과 모든 리다이렉트 홉이 사설 및 예약 대역(클라우드 메타데이터 포함)에 대해 검사되며, TCP 연결은 그 검사를 통과한 IP에 고정됩니다.
 
 ## 설치
 

docs/edge/ko/tools/web-scraping/scrapewebsitetool.mdx

@@ -16,6 +16,8 @@ mode: "wide"
 지정된 웹사이트의 내용을 추출하고 읽을 수 있도록 설계된 도구입니다. 이 도구는 HTTP 요청을 보내고 수신된 HTML 콘텐츠를 파싱함으로써 다양한 유형의 웹 페이지를 처리할 수 있습니다.
 이 도구는 웹 스크래핑 작업, 데이터 수집 또는 웹사이트에서 특정 정보를 추출하는 데 특히 유용할 수 있습니다.
 
+가져오기는 CrewAI의 SSRF 안전 HTTP 헬퍼를 거칩니다. 요청된 URL과 모든 리다이렉트 홉이 사설 및 예약 대역(클라우드 메타데이터 포함)에 대해 검사되며, TCP 연결은 그 검사를 통과한 IP에 고정됩니다.
+
 ## 설치
 
 crewai_tools 패키지를 설치하세요

docs/edge/pt-BR/tools/web-scraping/scrapeelementfromwebsitetool.mdx

@@ -9,7 +9,7 @@ mode: "wide"
 
 ## Descrição
 
-A `ScrapeElementFromWebsiteTool` foi projetada para extrair elementos específicos de websites utilizando seletores CSS. Esta ferramenta permite que agentes CrewAI capturem conteúdos direcionados de páginas web, tornando-se útil para tarefas de extração de dados em que apenas partes específicas de uma página são necessárias.
+A `ScrapeElementFromWebsiteTool` foi projetada para extrair elementos específicos de websites utilizando seletores CSS. Esta ferramenta permite que agentes CrewAI capturem conteúdos direcionados de páginas web, tornando-se útil para tarefas de extração de dados em que apenas partes específicas de uma página são necessárias. As buscas passam pelo helper HTTP seguro contra SSRF do CrewAI: a URL solicitada e cada hop de redirecionamento são verificados contra faixas privadas e reservadas (incluindo metadados de nuvem), e a conexão TCP é fixada no IP que passou nessa verificação.
 
 ## Instalação
 

docs/edge/pt-BR/tools/web-scraping/scrapewebsitetool.mdx

@@ -16,6 +16,8 @@ mode: "wide"
 Uma ferramenta desenvolvida para extrair e ler o conteúdo de um site especificado. Ela é capaz de lidar com diversos tipos de páginas web fazendo requisições HTTP e analisando o conteúdo HTML recebido.
 Esta ferramenta pode ser especialmente útil para tarefas de raspagem de dados, coleta de dados ou extração de informações específicas de sites.
 
+As buscas passam pelo helper HTTP seguro contra SSRF do CrewAI: a URL solicitada e cada hop de redirecionamento são verificados contra faixas privadas e reservadas (incluindo metadados de nuvem), e a conexão TCP é fixada no IP que passou nessa verificação.
+
 ## Instalação
 
 Instale o pacote crewai_tools

lib/crewai-tools/src/crewai_tools/rag/loaders/docs_site_loader.py

@@ -29,7 +29,7 @@ class DocsSiteLoader(BaseLoader):
         try:
             response = safe_get(docs_url, timeout=30)
             response.raise_for_status()
-        except requests.RequestException as e:
+        except (requests.RequestException, ValueError) as e:
             raise ValueError(
                 f"Unable to fetch documentation from {docs_url}: {e}"
             ) from e

lib/crewai-tools/src/crewai_tools/security/safe_path.py

@@ -5,7 +5,9 @@ file access and server-side request forgery (SSRF) when tools accept
 user-controlled or LLM-controlled inputs at runtime.
 
 Set CREWAI_TOOLS_ALLOW_UNSAFE_PATHS=true to bypass validation (not
-recommended for production).
+recommended for production). Managed workers should set
+CREWAI_TOOLS_FORCE_SAFE_PATHS=true so a tenant cannot disable these
+checks by exporting the escape hatch on their own deployment.
 """
 
 from __future__ import annotations
@@ -20,6 +22,7 @@ from urllib.parse import urlparse
 logger = logging.getLogger(__name__)
 
 _UNSAFE_PATHS_ENV = "CREWAI_TOOLS_ALLOW_UNSAFE_PATHS"
+_FORCE_SAFE_PATHS_ENV = "CREWAI_TOOLS_FORCE_SAFE_PATHS"
 _BYPASS_HINT = f"Set {_UNSAFE_PATHS_ENV}=true to bypass this check."
 
 
@@ -69,9 +72,21 @@ def format_sandbox_error(error: Exception, remedy: str) -> str:
     return f"{text} {remedy}".strip()
 
 
+def _env_flag_enabled(name: str) -> bool:
+    return os.environ.get(name, "").lower() in ("true", "1", "yes")
+
+
 def _is_escape_hatch_enabled() -> bool:
-    """Check if the unsafe paths escape hatch is enabled."""
-    return os.environ.get(_UNSAFE_PATHS_ENV, "").lower() in ("true", "1", "yes")
+    """True when ``ALLOW_UNSAFE_PATHS`` is set and ``FORCE_SAFE_PATHS`` is not."""
+    if _env_flag_enabled(_FORCE_SAFE_PATHS_ENV):
+        if _env_flag_enabled(_UNSAFE_PATHS_ENV):
+            logger.warning(
+                "%s is set; ignoring %s",
+                _FORCE_SAFE_PATHS_ENV,
+                _UNSAFE_PATHS_ENV,
+            )
+        return False
+    return _env_flag_enabled(_UNSAFE_PATHS_ENV)
 
 
 def validate_file_path(path: str, base_dir: str | None = None) -> str:
@@ -161,8 +176,8 @@ _BLOCKED_IPV6_NETWORKS = [
 ]
 
 
-def _is_private_or_reserved(ip_str: str) -> bool:
-    """Check if an IP address is private, reserved, or otherwise unsafe."""
+def is_blocked_ip(ip_str: str) -> bool:
+    """Return True if *ip_str* is private, reserved, or otherwise unsafe to fetch."""
     try:
         addr = ipaddress.ip_address(ip_str)
         # Unwrap IPv4-mapped IPv6 addresses (e.g., ::ffff:127.0.0.1) to IPv4
@@ -187,6 +202,9 @@ def validate_url(url: str) -> str:
     DNS and checks that the target IP is not private or reserved (prevents
     SSRF to internal services and cloud metadata endpoints).
 
+    This checks the URL string only. Fetch with ``safe_get`` so the
+    connection is pinned to an authorised IP.
+
     Args:
         url: The URL to validate.
 
@@ -232,7 +250,7 @@ def validate_url(url: str) -> str:
 
     for _family, _, _, _, sockaddr in addrinfos:
         ip_str = str(sockaddr[0])
-        if _is_private_or_reserved(ip_str):
+        if is_blocked_ip(ip_str):
             raise ValueError(
                 f"URL '{url}' resolves to private/reserved IP {ip_str}. "
                 f"Access to internal networks is not allowed. "

lib/crewai-tools/src/crewai_tools/security/safe_requests.py

@@ -1,4 +1,8 @@
-"""HTTP helpers that preserve crewai-tools URL safety checks."""
+"""SSRF-safe HTTP fetching for crewai-tools.
+
+``safe_get`` validates each URL and redirect hop, then fetches through a
+session that pins TCP to the checked IP and ignores environment proxies.
+"""
 
 from __future__ import annotations
 
@@ -7,7 +11,12 @@ from urllib.parse import urljoin, urlparse
 
 import requests
 
-from crewai_tools.security.safe_path import validate_url
+from crewai_tools.security.safe_path import (
+    _BYPASS_HINT,
+    _is_escape_hatch_enabled,
+    validate_url,
+)
+from crewai_tools.security.ssrf_adapter import SSRFProtectedAdapter
 
 
 _REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
@@ -49,6 +58,70 @@ def _strip_cross_origin_credentials(request_kwargs: dict[str, Any]) -> dict[str,
     return sanitized
 
 
+class _SafeSession(requests.Session):
+    """Session that does not follow redirects unless the caller opts in."""
+
+    def get(self, url: str | bytes, **kwargs: Any) -> requests.Response:
+        kwargs.setdefault("allow_redirects", False)
+        return super().get(url, **kwargs)
+
+    def options(self, url: str | bytes, **kwargs: Any) -> requests.Response:
+        kwargs.setdefault("allow_redirects", False)
+        return super().options(url, **kwargs)
+
+    def request(
+        self, method: str | bytes, url: str | bytes, *args: Any, **kwargs: Any
+    ) -> requests.Response:
+        kwargs.setdefault("allow_redirects", False)
+        return super().request(method, url, *args, **kwargs)
+
+
+def create_safe_session() -> requests.Session:
+    """Return a session that pins TCP to a validated peer and does not follow redirects."""
+    session = _SafeSession()
+    session.trust_env = False
+    session.proxies = {}
+    adapter = SSRFProtectedAdapter()
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    return session
+
+
+def _reject_proxies(kwargs: dict[str, Any]) -> None:
+    proxies = kwargs.pop("proxies", None)
+    if proxies and not _is_escape_hatch_enabled():
+        raise ValueError(f"Proxies are not allowed for safe_get. {_BYPASS_HINT}")
+    kwargs["proxies"] = {}
+
+
+def _attach_session(response: requests.Response, session: requests.Session) -> None:
+    """Keep *session* alive until *response* is closed (needed for ``stream=True``)."""
+    original_close = response.close
+
+    def close_with_session() -> None:
+        try:
+            original_close()
+        finally:
+            session.close()
+
+    response.close = close_with_session  # type: ignore[method-assign]
+
+
+def _raw_get(url: str, **kwargs: Any) -> requests.Response:
+    """GET through an SSRF-protected session."""
+    session = create_safe_session()
+    owns_session = True
+    try:
+        response = session.get(url, **kwargs)
+        if kwargs.get("stream"):
+            _attach_session(response, session)
+            owns_session = False
+        return response
+    finally:
+        if owns_session:
+            session.close()
+
+
 def safe_get(url: str, *, max_redirects: int = 10, **kwargs: Any) -> requests.Response:
     """GET a URL while validating each redirect target before following it.
 
@@ -58,6 +131,7 @@ def safe_get(url: str, *, max_redirects: int = 10, **kwargs: Any) -> requests.Re
     until its body is read or closed.
     """
     current_url = validate_url(url)
+    _reject_proxies(kwargs)
     request_kwargs = {**kwargs, "allow_redirects": False}
     timeout = request_kwargs.pop("timeout", 30)
     history: list[requests.Response] = []
@@ -65,7 +139,7 @@ def safe_get(url: str, *, max_redirects: int = 10, **kwargs: Any) -> requests.Re
 
     try:
         while True:
-            response = requests.get(current_url, timeout=timeout, **request_kwargs)
+            response = _raw_get(current_url, timeout=timeout, **request_kwargs)
             if (
                 response.status_code not in _REDIRECT_STATUS_CODES
                 or "Location" not in response.headers

lib/crewai-tools/src/crewai_tools/security/ssrf_adapter.py

@@ -0,0 +1,242 @@
+"""urllib3/requests transport that pins TCP to a validated peer IP."""
+
+from __future__ import annotations
+
+import socket
+import sys
+from typing import Any
+
+import requests
+from requests.adapters import DEFAULT_POOLBLOCK, HTTPAdapter
+from urllib3.connection import HTTPConnection, HTTPSConnection
+from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
+from urllib3.exceptions import (
+    ConnectTimeoutError,
+    LocationParseError,
+    NameResolutionError,
+    NewConnectionError,
+)
+from urllib3.poolmanager import PoolManager
+from urllib3.util.connection import (
+    allowed_gai_family,
+    create_connection as urllib3_create_connection,
+)
+
+from crewai_tools.security.safe_path import (
+    _BYPASS_HINT,
+    _is_escape_hatch_enabled,
+    is_blocked_ip,
+)
+
+
+def _set_socket_options(sock: socket.socket, options: Any) -> None:
+    for opt in options or ():
+        sock.setsockopt(*opt)
+
+
+def _connect_timeout(timeout: Any) -> float | None:
+    connect = getattr(timeout, "connect_timeout", timeout)
+    if connect is None or connect is getattr(timeout, "DEFAULT_TIMEOUT", None):
+        return socket.getdefaulttimeout()
+    if isinstance(connect, (int, float)):
+        return float(connect)
+    return socket.getdefaulttimeout()
+
+
+def _blocked_ip_error(ip_str: str) -> ValueError:
+    return ValueError(
+        f"Connection resolved to private/reserved IP {ip_str}. "
+        f"Access to internal networks is not allowed (possible SSRF via "
+        f"redirect or DNS rebinding). {_BYPASS_HINT}"
+    )
+
+
+def _assert_safe_peer(sock: socket.socket) -> None:
+    """Raise if a connected socket's peer is a private/reserved address."""
+    if _is_escape_hatch_enabled():
+        return
+    try:
+        peer = sock.getpeername()
+    except OSError as exc:
+        raise ValueError(
+            "Unable to determine the connected peer address; blocking "
+            f"request to prevent SSRF. {_BYPASS_HINT}"
+        ) from exc
+    ip_str = str(peer[0])
+    if is_blocked_ip(ip_str):
+        raise _blocked_ip_error(ip_str)
+
+
+def create_validated_connection(
+    host: str,
+    port: int,
+    *,
+    timeout: Any = None,
+    source_address: tuple[str, int] | None = None,
+    socket_options: Any = None,
+) -> socket.socket:
+    """Open a TCP socket to *host* after validating and pinning the peer IP."""
+    if _is_escape_hatch_enabled():
+        return urllib3_create_connection(
+            (host, port),
+            timeout=_connect_timeout(timeout),
+            source_address=source_address,
+            socket_options=socket_options,
+        )
+
+    if host.startswith("["):
+        host = host.strip("[]")
+
+    try:
+        host.encode("idna")
+    except UnicodeError:
+        raise LocationParseError(f"'{host}', label empty or too long") from None
+
+    try:
+        addrinfos = socket.getaddrinfo(
+            host, port, allowed_gai_family(), socket.SOCK_STREAM
+        )
+    except socket.gaierror as exc:
+        raise ValueError(f"Could not resolve hostname: '{host}'") from exc
+
+    for _family, _socktype, _proto, _canonname, sockaddr in addrinfos:
+        ip_str = str(sockaddr[0])
+        if is_blocked_ip(ip_str):
+            raise _blocked_ip_error(ip_str)
+
+    timeout = _connect_timeout(timeout)
+    err: OSError | None = None
+    for family, socktype, proto, _canonname, sockaddr in addrinfos:
+        sock: socket.socket | None = None
+        try:
+            sock = socket.socket(family, socktype, proto)
+            _set_socket_options(sock, socket_options)
+            sock.settimeout(timeout)
+            if source_address:
+                sock.bind(source_address)
+            sock.connect(sockaddr)
+            peer_validated = False
+            try:
+                _assert_safe_peer(sock)
+                peer_validated = True
+            finally:
+                if not peer_validated:
+                    sock.close()
+            return sock
+        except OSError as exc:
+            err = exc
+            if sock is not None:
+                sock.close()
+
+    if err is not None:
+        raise err
+    raise OSError("getaddrinfo returns an empty list")
+
+
+def _open_validated_socket(conn: HTTPConnection) -> socket.socket:
+    port = conn.port
+    if port is None:
+        port = 443 if isinstance(conn, HTTPSConnection) else 80
+    try:
+        sock = create_validated_connection(
+            conn._dns_host,
+            port,
+            timeout=conn.timeout,
+            source_address=conn.source_address,
+            socket_options=conn.socket_options,
+        )
+    except socket.gaierror as exc:
+        raise NameResolutionError(conn.host, conn, exc) from exc
+    except socket.timeout as exc:
+        raise ConnectTimeoutError(
+            conn,
+            f"Connection to {conn.host} timed out. (connect timeout={conn.timeout})",
+        ) from exc
+    except OSError as exc:
+        raise NewConnectionError(
+            conn, f"Failed to establish a new connection: {exc}"
+        ) from exc
+
+    sys.audit("http.client.connect", conn, conn.host, conn.port)
+    return sock
+
+
+class _SafeHTTPConnection(HTTPConnection):
+    def _new_conn(self) -> socket.socket:
+        return _open_validated_socket(self)
+
+
+class _SafeHTTPSConnection(HTTPSConnection):
+    def _new_conn(self) -> socket.socket:
+        return _open_validated_socket(self)
+
+
+class _SafeHTTPConnectionPool(HTTPConnectionPool):
+    ConnectionCls = _SafeHTTPConnection
+
+
+class _SafeHTTPSConnectionPool(HTTPSConnectionPool):
+    ConnectionCls = _SafeHTTPSConnection
+
+
+_SAFE_POOL_CLASSES = {
+    "http": _SafeHTTPConnectionPool,
+    "https": _SafeHTTPSConnectionPool,
+}
+
+
+class _SafePoolManager(PoolManager):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.pool_classes_by_scheme = _SAFE_POOL_CLASSES
+
+
+class SSRFProtectedAdapter(HTTPAdapter):
+    """HTTPAdapter that connects through :func:`create_validated_connection`."""
+
+    def init_poolmanager(
+        self,
+        connections: int,
+        maxsize: int,
+        block: bool = DEFAULT_POOLBLOCK,
+        **pool_kwargs: Any,
+    ) -> None:
+        self._pool_connections = connections
+        self._pool_maxsize = maxsize
+        self._pool_block = block
+        self.poolmanager = _SafePoolManager(
+            num_pools=connections,
+            maxsize=maxsize,
+            block=block,
+            **pool_kwargs,
+        )
+
+    def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any:
+        if not _is_escape_hatch_enabled():
+            raise ValueError(
+                f"Proxies are not allowed for SSRF-safe requests. {_BYPASS_HINT}"
+            )
+        return super().proxy_manager_for(proxy, **proxy_kwargs)  # type: ignore[no-untyped-call]
+
+    def send(
+        self,
+        request: requests.PreparedRequest,
+        stream: bool = False,
+        timeout: Any = None,
+        verify: bool | str = True,
+        cert: Any = None,
+        proxies: Any = None,
+    ) -> requests.Response:
+        unsafe = _is_escape_hatch_enabled()
+        if proxies and not unsafe:
+            raise ValueError(
+                f"Proxies are not allowed for SSRF-safe requests. {_BYPASS_HINT}"
+            )
+        return super().send(
+            request,
+            stream=stream,
+            timeout=timeout,
+            verify=verify,
+            cert=cert,
+            proxies=proxies if unsafe else {},
+        )

lib/crewai-tools/src/crewai_tools/tools/url_read_tool/url_read_tool.py

@@ -114,14 +114,12 @@ class URLReadTool(BaseTool):
         automatically: every hop is revalidated, and credentials are dropped on
         cross-origin hops. Bodies over ``max_bytes`` are abandoned mid-stream.
 
-        Two risks are not closed here. Validation resolves the hostname and
-        requests resolves it again when connecting, so a DNS entry that changes
-        between those lookups can still redirect the connection (DNS
-        rebinding); closing that requires pinning the connection to the
-        validated address. And the returned text is untrusted remote content
-        flowing into an agent's context -- a fetched page can attempt to
-        instruct the agent. Neither is addressable by input validation alone;
-        network egress policy and prompt-level handling cover them.
+        The fetch pins TCP to the IP that passed validation unless
+        ``CREWAI_TOOLS_ALLOW_UNSAFE_PATHS`` is set without
+        ``CREWAI_TOOLS_FORCE_SAFE_PATHS``. The returned text is still
+        untrusted remote content flowing into an agent's context -- a
+        fetched page can attempt to instruct the agent. Network egress
+        policy and prompt-level handling cover that.
 
     Args:
         max_bytes (int): Largest response body to accept, in decoded bytes.

lib/crewai-tools/tests/rag/test_csv_loader.py

@@ -85,7 +85,7 @@ class TestCSVLoader:
 
         assert result1.doc_id == result2.doc_id
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_csv_from_url(self, mock_get):
         mock_get.return_value = Mock(
             text="name,value\ntest,123", raise_for_status=Mock(return_value=None)
@@ -99,7 +99,7 @@ class TestCSVLoader:
         assert "text/csv" in headers["Accept"]
         assert "crewai-tools CSVLoader" in headers["User-Agent"]
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_csv_with_custom_headers(self, mock_get):
         mock_get.return_value = Mock(
             text="data,value\ntest,456", raise_for_status=Mock(return_value=None)
@@ -112,7 +112,7 @@ class TestCSVLoader:
         assert "Headers: data | value" in result.content
         assert mock_get.call_args[1]["headers"] == headers
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_csv_loader_handles_network_errors(self, mock_get):
         mock_get.side_effect = Exception("Network error")
         loader = CSVLoader()
@@ -120,7 +120,7 @@ class TestCSVLoader:
         with pytest.raises(ValueError, match="Error fetching content from URL"):
             loader.load(SourceContent("https://example.com/data.csv"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_csv_loader_handles_http_error(self, mock_get):
         mock_get.return_value = Mock()
         mock_get.return_value.raise_for_status.side_effect = Exception("404 Not Found")

lib/crewai-tools/tests/rag/test_docx_loader.py

@@ -41,7 +41,7 @@ class TestDOCXLoader:
 
             assert result.metadata["tables"] == 2
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("docx.Document")
     @patch("tempfile.NamedTemporaryFile")
     @patch("os.unlink")
@@ -77,7 +77,7 @@ class TestDOCXLoader:
 
         mock_temp.write.assert_called_once_with(b"fake docx content")
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("docx.Document")
     def test_load_docx_from_url_with_custom_headers(self, mock_docx_class, mock_get):
         mock_get.return_value = Mock(
@@ -95,7 +95,7 @@ class TestDOCXLoader:
 
         assert mock_get.call_args[1]["headers"] == custom_headers
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_docx_url_download_error(self, mock_get):
         mock_get.side_effect = Exception("Network error")
 
@@ -103,7 +103,7 @@ class TestDOCXLoader:
         with pytest.raises(ValueError, match="Error fetching content from URL"):
             loader.load(SourceContent("https://example.com/test.docx"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_docx_url_http_error(self, mock_get):
         mock_get.return_value = Mock(
             raise_for_status=Mock(side_effect=Exception("404 Not Found"))

lib/crewai-tools/tests/rag/test_json_loader.py

@@ -127,7 +127,7 @@ class TestJSONLoader:
             os.unlink(path)
 
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_url_response_valid_json(self, mock_get):
         mock_get.return_value = Mock(
             text='{"key": "value", "number": 123}',
@@ -143,7 +143,7 @@ class TestJSONLoader:
         assert "application/json" in headers["Accept"]
         assert "crewai-tools JSONLoader" in headers["User-Agent"]
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_url_response_not_json(self, mock_get):
         mock_get.return_value = Mock(
             text='{"key": "value"}',
@@ -155,7 +155,7 @@ class TestJSONLoader:
         result = loader.load(SourceContent("https://example.com/data.json"))
         assert all(part in result.content for part in ["key", "value"])
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_url_with_custom_headers(self, mock_get):
         mock_get.return_value = Mock(
             text='{"data": "test"}',
@@ -169,14 +169,14 @@ class TestJSONLoader:
 
         assert mock_get.call_args[1]["headers"] == headers
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_url_network_failure(self, mock_get):
         mock_get.side_effect = Exception("Network error")
         loader = JSONLoader()
         with pytest.raises(ValueError, match="Error fetching content from URL"):
             loader.load(SourceContent("https://api.example.com/data.json"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_url_http_error(self, mock_get):
         mock_get.return_value = Mock(
             raise_for_status=Mock(side_effect=Exception("404"))

lib/crewai-tools/tests/rag/test_mdx_loader.py

@@ -138,7 +138,7 @@ Final content.
         assert "Only JSX content" in result.content
         assert "No markdown here" in result.content
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_mdx_from_url(self, mock_get):
         mock_get.return_value = Mock(
             text="# MDX from URL\n\nContent here.\n\n<Component />",
@@ -149,7 +149,7 @@ Final content.
         assert "# MDX from URL" in result.content
         assert "<Component />" not in result.content
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_load_mdx_with_custom_headers(self, mock_get):
         mock_get.return_value = Mock(
             text="# Custom headers test", raise_for_status=lambda: None
@@ -161,7 +161,7 @@ Final content.
         )
         assert mock_get.call_args[1]["headers"] == {"Authorization": "Bearer token"}
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_mdx_url_fetch_error(self, mock_get):
         mock_get.side_effect = Exception("Network error")
         with pytest.raises(ValueError, match="Error fetching content from URL https://example.com: Network error"):

lib/crewai-tools/tests/rag/test_webpage_loader.py

@@ -22,7 +22,7 @@ class TestWebPageLoader:
         soup.return_value = script_style_elements or []
         return soup
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_load_basic_webpage(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -37,7 +37,7 @@ class TestWebPageLoader:
         assert result.content == "Test content"
         assert result.metadata["title"] == "Test Page"
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_load_webpage_with_scripts_and_styles(self, mock_bs, mock_get):
         html = """
@@ -62,7 +62,7 @@ class TestWebPageLoader:
         for el in scripts + styles:
             el.decompose.assert_called_once()
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_text_cleaning_and_title_handling(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -77,7 +77,7 @@ class TestWebPageLoader:
         assert result.content is not None
         assert result.metadata["title"] == ""
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_empty_or_missing_title(self, mock_bs, mock_get):
         for title in [None, ""]:
@@ -90,7 +90,7 @@ class TestWebPageLoader:
             result = loader.load(SourceContent("https://example.com"))
             assert result.metadata["title"] == ""
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_custom_and_default_headers(self, mock_get):
         mock_get.return_value = self.setup_mock_response(
             "<html><body>Test</body></html>"
@@ -109,14 +109,14 @@ class TestWebPageLoader:
 
         assert mock_get.call_args[1]["headers"] == custom_headers
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_error_handling(self, mock_get):
         for error in [Exception("Fail"), ValueError("Bad"), ImportError("Oops")]:
             mock_get.side_effect = error
             with pytest.raises(ValueError, match="Error loading webpage"):
                 WebPageLoader().load(SourceContent("https://example.com"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_timeout_and_http_error(self, mock_get):
         import requests
 
@@ -131,7 +131,7 @@ class TestWebPageLoader:
         with pytest.raises(ValueError):
             WebPageLoader().load(SourceContent("https://example.com/404"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_doc_id_consistency(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -145,7 +145,7 @@ class TestWebPageLoader:
 
         assert result1.doc_id == result2.doc_id
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_status_code_and_content_type(self, mock_bs, mock_get):
         for status in [200, 201, 301]:

lib/crewai-tools/tests/rag/test_xml_loader.py

@@ -22,7 +22,7 @@ class TestWebPageLoader:
         soup.return_value = script_style_elements or []
         return soup
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_load_basic_webpage(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -37,7 +37,7 @@ class TestWebPageLoader:
         assert result.content == "Test content"
         assert result.metadata["title"] == "Test Page"
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_load_webpage_with_scripts_and_styles(self, mock_bs, mock_get):
         html = """
@@ -62,7 +62,7 @@ class TestWebPageLoader:
         for el in scripts + styles:
             el.decompose.assert_called_once()
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_text_cleaning_and_title_handling(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -77,7 +77,7 @@ class TestWebPageLoader:
         assert result.content is not None
         assert result.metadata["title"] == ""
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_empty_or_missing_title(self, mock_bs, mock_get):
         for title in [None, ""]:
@@ -90,7 +90,7 @@ class TestWebPageLoader:
             result = loader.load(SourceContent("https://example.com"))
             assert result.metadata["title"] == ""
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_custom_and_default_headers(self, mock_get):
         mock_get.return_value = self.setup_mock_response(
             "<html><body>Test</body></html>"
@@ -109,14 +109,14 @@ class TestWebPageLoader:
 
         assert mock_get.call_args[1]["headers"] == custom_headers
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_error_handling(self, mock_get):
         for error in [Exception("Fail"), ValueError("Bad"), ImportError("Oops")]:
             mock_get.side_effect = error
             with pytest.raises(ValueError, match="Error loading webpage"):
                 WebPageLoader().load(SourceContent("https://example.com"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     def test_timeout_and_http_error(self, mock_get):
         import requests
 
@@ -131,7 +131,7 @@ class TestWebPageLoader:
         with pytest.raises(ValueError):
             WebPageLoader().load(SourceContent("https://example.com/404"))
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_doc_id_consistency(self, mock_bs, mock_get):
         mock_get.return_value = self.setup_mock_response(
@@ -145,7 +145,7 @@ class TestWebPageLoader:
 
         assert result1.doc_id == result2.doc_id
 
-    @patch("requests.get")
+    @patch("crewai_tools.security.safe_requests._raw_get")
     @patch("crewai_tools.rag.loaders.webpage_loader.BeautifulSoup")
     def test_status_code_and_content_type(self, mock_bs, mock_get):
         for status in [200, 201, 301]:

lib/crewai-tools/tests/utilities/test_safe_path.py

@@ -193,6 +193,13 @@ class TestValidateUrl:
         result = validate_url("file:///etc/passwd")
         assert result == "file:///etc/passwd"
 
+    def test_force_safe_paths_overrides_escape_hatch(self, monkeypatch):
+        """Managed workers can ignore a tenant-supplied escape hatch."""
+        monkeypatch.setenv("CREWAI_TOOLS_ALLOW_UNSAFE_PATHS", "true")
+        monkeypatch.setenv("CREWAI_TOOLS_FORCE_SAFE_PATHS", "true")
+        with pytest.raises(ValueError, match="private/reserved IP"):
+            validate_url("http://127.0.0.1/admin")
+
 
 class TestFormatSandboxError:
     def test_replaces_bypass_advice_with_remedy(self, tmp_path):

lib/crewai-tools/tests/utilities/test_safe_requests.py

@@ -1,4 +1,4 @@
-"""Tests for redirect-aware safe HTTP helpers."""
+"""Tests for redirect-aware, connection-pinning safe HTTP helpers."""
 
 from __future__ import annotations
 
@@ -9,7 +9,16 @@ from typing import Any
 import pytest
 import requests
 
-from crewai_tools.security.safe_requests import safe_get
+from crewai_tools.security import safe_requests
+from crewai_tools.security.safe_requests import (
+    SSRFProtectedAdapter,
+    create_safe_session,
+    safe_get,
+)
+from crewai_tools.security.ssrf_adapter import (
+    _assert_safe_peer,
+    create_validated_connection,
+)
 
 
 def _response(url: str, status_code: int, *, location: str | None = None) -> requests.Response:
@@ -52,7 +61,7 @@ def test_safe_get_blocks_direct_internal_url() -> None:
 
 def _mock_get(monkeypatch: pytest.MonkeyPatch, get_response: Any) -> None:
     monkeypatch.setattr(
-        "crewai_tools.security.safe_requests.requests.get",
+        "crewai_tools.security.safe_requests._raw_get",
         get_response,
     )
 
@@ -249,3 +258,336 @@ def test_safe_get_preserves_credentials_on_same_origin_redirect(
 
     assert requests_made[1][1]["headers"] == {"Authorization": "Bearer token"}
     assert requests_made[1][1]["cookies"] == {"session": "abc"}
+
+
+def test_safe_get_rejects_proxies(
+    monkeypatch: pytest.MonkeyPatch, public_dns: None
+) -> None:
+    _mock_get(monkeypatch, lambda url, **kwargs: _response(url, 200))
+
+    with pytest.raises(ValueError, match="Proxies are not allowed"):
+        safe_get(
+            "http://public.example/start",
+            timeout=15,
+            proxies={"http": "http://127.0.0.1:8080"},
+        )
+
+
+def test_session_mounts_protected_adapter_and_ignores_env_proxies() -> None:
+    session = create_safe_session()
+    assert isinstance(session.get_adapter("http://x"), SSRFProtectedAdapter)
+    assert isinstance(session.get_adapter("https://x"), SSRFProtectedAdapter)
+    assert session.trust_env is False
+    assert session.proxies == {}
+
+
+def test_safe_session_does_not_follow_redirects_by_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session = create_safe_session()
+    urls: list[str] = []
+
+    def fake_send(request: requests.PreparedRequest, **kwargs: Any) -> requests.Response:
+        urls.append(request.url or "")
+        response = requests.Response()
+        response.status_code = 302
+        response.url = request.url
+        response.request = request
+        response.headers["Location"] = "http://127.0.0.1/admin"
+        response._content = b""
+        response.raw = BytesIO()
+        return response
+
+    adapter = session.get_adapter("http://example.com/")
+    monkeypatch.setattr(adapter, "send", fake_send)
+
+    response = session.get("http://example.com/start")
+
+    assert response.status_code == 302
+    assert urls == ["http://example.com/start"]
+
+
+def test_safe_session_follows_redirects_when_caller_opts_in(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session = create_safe_session()
+    urls: list[str] = []
+
+    def fake_send(request: requests.PreparedRequest, **kwargs: Any) -> requests.Response:
+        url = request.url or ""
+        urls.append(url)
+        response = requests.Response()
+        response.url = url
+        response.request = request
+        response._content = b""
+        response.raw = BytesIO()
+        if url.endswith("/start"):
+            response.status_code = 302
+            response.headers["Location"] = "http://example.com/final"
+        else:
+            response.status_code = 200
+        return response
+
+    adapter = session.get_adapter("http://example.com/")
+    monkeypatch.setattr(adapter, "send", fake_send)
+
+    response = session.get("http://example.com/start", allow_redirects=True)
+
+    assert response.status_code == 200
+    assert urls == ["http://example.com/start", "http://example.com/final"]
+
+
+def test_adapter_rejects_proxies() -> None:
+    adapter = SSRFProtectedAdapter()
+    req = requests.Request("GET", "http://example.com/").prepare()
+    with pytest.raises(ValueError, match="Proxies are not allowed"):
+        adapter.send(req, proxies={"http": "http://127.0.0.1:8080"})
+
+
+class _FakeSock:
+    def __init__(self, peer: tuple[str, int]) -> None:
+        self._peer = peer
+
+    def getpeername(self) -> tuple[str, int]:
+        return self._peer
+
+
+def test_assert_safe_peer_blocks_private() -> None:
+    with pytest.raises(ValueError, match="private/reserved"):
+        _assert_safe_peer(_FakeSock(("127.0.0.1", 80)))
+
+
+def test_assert_safe_peer_blocks_metadata() -> None:
+    with pytest.raises(ValueError, match="private/reserved"):
+        _assert_safe_peer(_FakeSock(("169.254.169.254", 80)))
+
+
+def test_assert_safe_peer_allows_public() -> None:
+    _assert_safe_peer(_FakeSock(("93.184.216.34", 80)))
+
+
+def test_assert_safe_peer_respects_escape_hatch(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("CREWAI_TOOLS_ALLOW_UNSAFE_PATHS", "true")
+    _assert_safe_peer(_FakeSock(("127.0.0.1", 80)))
+
+
+def test_assert_safe_peer_force_safe_overrides_escape_hatch(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("CREWAI_TOOLS_ALLOW_UNSAFE_PATHS", "true")
+    monkeypatch.setenv("CREWAI_TOOLS_FORCE_SAFE_PATHS", "true")
+    with pytest.raises(ValueError, match="private/reserved"):
+        _assert_safe_peer(_FakeSock(("127.0.0.1", 80)))
+
+
+def test_create_validated_connection_pins_resolved_ip(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    lookups = {"n": 0}
+    connected_to: list[tuple[str, int]] = []
+
+    def fake_getaddrinfo(
+        host: str, port: int, *args: Any, **kwargs: Any
+    ) -> list[tuple[Any, ...]]:
+        lookups["n"] += 1
+        ip = "93.184.216.34" if lookups["n"] == 1 else "169.254.169.254"
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, port or 80)),
+        ]
+
+    class RecordingSocket:
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            self.peer: tuple[str, int] | None = None
+
+        def setsockopt(self, *args: Any, **kwargs: Any) -> None:
+            return None
+
+        def settimeout(self, timeout: Any) -> None:
+            return None
+
+        def connect(self, sockaddr: tuple[str, int]) -> None:
+            connected_to.append(sockaddr)
+            self.peer = sockaddr
+
+        def getpeername(self) -> tuple[str, int]:
+            assert self.peer is not None
+            return self.peer
+
+        def close(self) -> None:
+            return None
+
+    monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
+    monkeypatch.setattr(socket, "socket", lambda *a, **k: RecordingSocket())
+
+    sock = create_validated_connection("rebind.example", 80)
+
+    assert connected_to == [("93.184.216.34", 80)]
+    assert sock.getpeername() == ("93.184.216.34", 80)
+
+
+def test_create_validated_connection_blocks_when_any_record_is_private(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_getaddrinfo(
+        host: str, port: int, *args: Any, **kwargs: Any
+    ) -> list[tuple[Any, ...]]:
+        return [
+            (
+                socket.AF_INET,
+                socket.SOCK_STREAM,
+                6,
+                "",
+                ("93.184.216.34", port or 80),
+            ),
+            (
+                socket.AF_INET,
+                socket.SOCK_STREAM,
+                6,
+                "",
+                ("169.254.169.254", port or 80),
+            ),
+        ]
+
+    monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
+
+    with pytest.raises(ValueError, match="169.254.169.254"):
+        create_validated_connection("dual.example", 80)
+
+
+def test_create_validated_connection_blocks_direct_loopback() -> None:
+    with pytest.raises(ValueError, match="private/reserved"):
+        create_validated_connection("127.0.0.1", 9)
+
+
+def test_create_validated_connection_keeps_socket_when_called_from_except(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    closed: list[bool] = []
+
+    def fake_getaddrinfo(
+        host: str, port: int, *args: Any, **kwargs: Any
+    ) -> list[tuple[Any, ...]]:
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port or 80)),
+        ]
+
+    class RecordingSocket:
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            self.peer: tuple[str, int] | None = None
+
+        def setsockopt(self, *args: Any, **kwargs: Any) -> None:
+            return None
+
+        def settimeout(self, timeout: Any) -> None:
+            return None
+
+        def connect(self, sockaddr: tuple[str, int]) -> None:
+            self.peer = sockaddr
+
+        def getpeername(self) -> tuple[str, int]:
+            assert self.peer is not None
+            return self.peer
+
+        def close(self) -> None:
+            closed.append(True)
+
+    monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
+    monkeypatch.setattr(socket, "socket", lambda *a, **k: RecordingSocket())
+
+    try:
+        raise RuntimeError("caller is handling an error")
+    except RuntimeError:
+        sock = create_validated_connection("public.example", 80)
+
+    assert closed == []
+    assert sock.getpeername() == ("93.184.216.34", 80)
+
+
+def test_create_validated_connection_closes_socket_when_peer_is_blocked(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    closed: list[bool] = []
+
+    def fake_getaddrinfo(
+        host: str, port: int, *args: Any, **kwargs: Any
+    ) -> list[tuple[Any, ...]]:
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port or 80)),
+        ]
+
+    class RecordingSocket:
+        def setsockopt(self, *args: Any, **kwargs: Any) -> None:
+            return None
+
+        def settimeout(self, timeout: Any) -> None:
+            return None
+
+        def connect(self, sockaddr: tuple[str, int]) -> None:
+            return None
+
+        def getpeername(self) -> tuple[str, int]:
+            return ("127.0.0.1", 80)
+
+        def close(self) -> None:
+            closed.append(True)
+
+    monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
+    monkeypatch.setattr(socket, "socket", lambda *a, **k: RecordingSocket())
+
+    with pytest.raises(ValueError, match="private/reserved"):
+        create_validated_connection("rebind.example", 80)
+
+    assert closed == [True]
+
+
+class _TrackingSession:
+    """Session stand-in that records whether it was closed too early."""
+
+    def __init__(self) -> None:
+        self.closed = False
+
+    def get(self, url: str, **kwargs: Any) -> requests.Response:
+        response = requests.Response()
+        response.status_code = 200
+        response.url = url
+        response._content = b"hello" if not kwargs.get("stream") else False
+        response.raw = BytesIO()
+
+        def iter_content(
+            chunk_size: int = 1, decode_unicode: bool = False
+        ) -> Any:
+            if self.closed:
+                raise RuntimeError("session already closed")
+            yield b"hello"
+
+        response.iter_content = iter_content  # type: ignore[method-assign]
+        return response
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def test_streamed_raw_get_keeps_session_open_until_response_close(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session = _TrackingSession()
+    monkeypatch.setattr(safe_requests, "create_safe_session", lambda: session)
+
+    response = safe_requests._raw_get("http://example.com/file", stream=True)
+
+    assert session.closed is False
+    assert b"".join(response.iter_content()) == b"hello"
+    response.close()
+    assert session.closed is True
+
+
+def test_non_streamed_raw_get_closes_session_before_return(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session = _TrackingSession()
+    monkeypatch.setattr(safe_requests, "create_safe_session", lambda: session)
+
+    response = safe_requests._raw_get("http://example.com/file")
+
+    assert session.closed is True
+    assert response.content == b"hello"

lib/crewai-tools/tool.specs.json

@@ -26937,7 +26937,7 @@
             "type": "string"
           }
         },
-        "description": "Read the content at an arbitrary URL and return it as text.\n\nUnlike :class:`~crewai_tools.tools.file_read_tool.file_read_tool.FileReadTool`,\nwhich is confined to the local filesystem, this tool performs network\nrequests to addresses the caller -- often an LLM -- chooses at runtime. It\nis a separate tool for exactly that reason: granting it is granting network\negress, and that should be a deliberate choice rather than a flag on a\nfilesystem tool.\n\nResponses are decoded to text according to their content type. PDF and DOCX\nbodies have their text extracted, HTML is stripped to visible text, and\ntext-shaped types (plain text, Markdown, JSON, XML, YAML, CSV) are decoded\nas-is. Any other type is refused rather than returned as base64, keeping\nthis tool's output text-only.\n\nSecurity:\n    Requests go through :func:`~crewai_tools.security.safe_requests.safe_get_bounded`,\n    which resolves each hostname and rejects it when any resolved address is\n    private, loopback, link-local, or otherwise reserved -- covering cloud\n    metadata endpoints and internal services. Redirects are never followed\n    automatically: every hop is revalidated, and credentials are dropped on\n    cross-origin hops. Bodies over ``max_bytes`` are abandoned mid-stream.\n\n    Two risks are not closed here. Validation resolves the hostname and\n    requests resolves it again when connecting, so a DNS entry that changes\n    between those lookups can still redirect the connection (DNS\n    rebinding); closing that requires pinning the connection to the\n    validated address. And the returned text is untrusted remote content\n    flowing into an agent's context -- a fetched page can attempt to\n    instruct the agent. Neither is addressable by input validation alone;\n    network egress policy and prompt-level handling cover them.\n\nArgs:\n    max_bytes (int): Largest response body to accept, in decoded bytes.\n        Defaults to 5 MiB.\n    timeout (float): Per-request timeout in seconds. Defaults to 30.\n    headers (Optional[dict[str, str]]): Extra request headers. Developer\n        supplied, not chosen by the model.\n    encoding (Optional[str]): Force a text encoding instead of honoring the\n        charset the server declares.\n    **kwargs: Additional keyword arguments passed to BaseTool.\n\nExample:\n    >>> tool = URLReadTool()\n    >>> content = tool.run(url=\"https://example.com/report.pdf\")\n    >>> head = tool.run(url=\"https://example.com/data.csv\", line_count=20)",
+        "description": "Read the content at an arbitrary URL and return it as text.\n\nUnlike :class:`~crewai_tools.tools.file_read_tool.file_read_tool.FileReadTool`,\nwhich is confined to the local filesystem, this tool performs network\nrequests to addresses the caller -- often an LLM -- chooses at runtime. It\nis a separate tool for exactly that reason: granting it is granting network\negress, and that should be a deliberate choice rather than a flag on a\nfilesystem tool.\n\nResponses are decoded to text according to their content type. PDF and DOCX\nbodies have their text extracted, HTML is stripped to visible text, and\ntext-shaped types (plain text, Markdown, JSON, XML, YAML, CSV) are decoded\nas-is. Any other type is refused rather than returned as base64, keeping\nthis tool's output text-only.\n\nSecurity:\n    Requests go through :func:`~crewai_tools.security.safe_requests.safe_get_bounded`,\n    which resolves each hostname and rejects it when any resolved address is\n    private, loopback, link-local, or otherwise reserved -- covering cloud\n    metadata endpoints and internal services. Redirects are never followed\n    automatically: every hop is revalidated, and credentials are dropped on\n    cross-origin hops. Bodies over ``max_bytes`` are abandoned mid-stream.\n\n    The fetch pins TCP to the IP that passed validation unless\n    ``CREWAI_TOOLS_ALLOW_UNSAFE_PATHS`` is set without\n    ``CREWAI_TOOLS_FORCE_SAFE_PATHS``. The returned text is still\n    untrusted remote content flowing into an agent's context -- a\n    fetched page can attempt to instruct the agent. Network egress\n    policy and prompt-level handling cover that.\n\nArgs:\n    max_bytes (int): Largest response body to accept, in decoded bytes.\n        Defaults to 5 MiB.\n    timeout (float): Per-request timeout in seconds. Defaults to 30.\n    headers (Optional[dict[str, str]]): Extra request headers. Developer\n        supplied, not chosen by the model.\n    encoding (Optional[str]): Force a text encoding instead of honoring the\n        charset the server declares.\n    **kwargs: Additional keyword arguments passed to BaseTool.\n\nExample:\n    >>> tool = URLReadTool()\n    >>> content = tool.run(url=\"https://example.com/report.pdf\")\n    >>> head = tool.run(url=\"https://example.com/data.csv\", line_count=20)",
         "properties": {
           "encoding": {
             "anyOf": [

lib/crewai/tests/llms/azure/test_azure_responses.py

@@ -10,6 +10,16 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+class _FakeOpenAICompletion:
+    """Plain stand-in: MagicMock is not reliably stored on Pydantic PrivateAttr."""
+
+    def __init__(self) -> None:
+        self.call = MagicMock(return_value="responses-result")
+        self.acall = AsyncMock(return_value="async-responses-result")
+        self.last_response_id = "resp_abc123"
+        self.last_reasoning_items = [{"type": "reasoning"}]
+        self.reset_chain = MagicMock()
+        self.reset_reasoning_chain = MagicMock()
 
 
 @pytest.fixture
@@ -32,13 +42,7 @@ def mock_openai_completion():
     Patches at the source module so that the dynamic import inside
     _init_responses_delegate picks up the mock.
     """
-    instance = MagicMock()
-    instance.call = MagicMock(return_value="responses-result")
-    instance.acall = AsyncMock(return_value="async-responses-result")
-    instance.last_response_id = "resp_abc123"
-    instance.last_reasoning_items = [{"type": "reasoning"}]
-    instance.reset_chain = MagicMock()
-    instance.reset_reasoning_chain = MagicMock()
+    instance = _FakeOpenAICompletion()
     mock_cls = MagicMock(return_value=instance)
 
     with patch(
@@ -245,8 +249,9 @@ class TestAzureResponsesProperties:
     """Test properties and methods delegated to the responses delegate."""
 
     def test_last_response_id(self, mock_openai_completion):
-        _mock_cls, _ = mock_openai_completion
+        _mock_cls, instance = mock_openai_completion
         comp = _create_azure_responses()
+        assert comp._responses_delegate is instance
         assert comp.last_response_id == "resp_abc123"
 
     def test_last_response_id_none_for_completions(self):
@@ -260,19 +265,22 @@ class TestAzureResponsesProperties:
         assert comp.last_response_id is None
 
     def test_last_reasoning_items(self, mock_openai_completion):
-        _mock_cls, _ = mock_openai_completion
+        _mock_cls, instance = mock_openai_completion
         comp = _create_azure_responses()
+        assert comp._responses_delegate is instance
         assert comp.last_reasoning_items == [{"type": "reasoning"}]
 
     def test_reset_chain(self, mock_openai_completion):
         _mock_cls, instance = mock_openai_completion
         comp = _create_azure_responses()
+        assert comp._responses_delegate is instance
         comp.reset_chain()
         instance.reset_chain.assert_called_once()
 
     def test_reset_reasoning_chain(self, mock_openai_completion):
         _mock_cls, instance = mock_openai_completion
         comp = _create_azure_responses()
+        assert comp._responses_delegate is instance
         comp.reset_reasoning_chain()
         instance.reset_reasoning_chain.assert_called_once()