fix(security): eliminate DNS rebinding TOCTOU and add SSRF validation to MCP tools

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

@@ -10,12 +10,16 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 import ipaddress
 import logging
 import os
 import socket
+from typing import Any
 from urllib.parse import urlparse
 
+from requests.adapters import HTTPAdapter
+
 
 logger = logging.getLogger(__name__)
 
@@ -84,10 +88,6 @@ def validate_file_path(path: str, base_dir: str | None = None) -> str:
         os.path.join(resolved_base, path) if not os.path.isabs(path) else path
     )
 
-    # Ensure the resolved path is within the base directory.
-    # When resolved_base already ends with a separator (e.g. the filesystem
-    # root "/"), appending os.sep would double it ("//"), so use the base
-    # as-is in that case.
     prefix = resolved_base if resolved_base.endswith(os.sep) else resolved_base + os.sep
     if not resolved_path.startswith(prefix) and resolved_path != resolved_base:
         raise ValueError(
@@ -143,9 +143,6 @@ def _is_private_or_reserved(ip_str: str) -> bool:
     """Check if an IP address is private, reserved, or otherwise unsafe."""
     try:
         addr = ipaddress.ip_address(ip_str)
-        # Unwrap IPv4-mapped IPv6 addresses (e.g., ::ffff:127.0.0.1) to IPv4
-        # so they are only checked against IPv4 networks (avoids TypeError when
-        # an IPv4Address is compared against an IPv6Network).
         if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped:
             addr = addr.ipv4_mapped
         networks = (
@@ -155,44 +152,34 @@ def _is_private_or_reserved(ip_str: str) -> bool:
         )
         return any(addr in network for network in networks)
     except ValueError:
-        return True  # If we can't parse, block it
-
+        return True
 
-def validate_url(url: str) -> str:
-    """Validate that a URL is safe to fetch.
-
-    Blocks ``file://`` scheme entirely. For ``http``/``https``, resolves
-    DNS and checks that the target IP is not private or reserved (prevents
-    SSRF to internal services and cloud metadata endpoints).
 
-    Args:
-        url: The URL to validate.
+def _validate_url_common(url: str) -> tuple[str, str, str | None]:
+    """Shared validation logic for URL schemes, hostnames, and DNS resolution.
 
     Returns:
-        The validated URL string.
+        Tuple of (scheme, hostname, resolved_ip) on success.
 
     Raises:
-        ValueError: If the URL uses a blocked scheme or resolves to a
-            private/reserved IP address.
+        ValueError: If the URL is unsafe.
     """
     if _is_escape_hatch_enabled():
         logger.warning(
             "%s is enabled — skipping URL validation for: %s",
             _UNSAFE_PATHS_ENV,
             url,
         )
-        return url
+        return "", "", None
 
     parsed = urlparse(url)
 
-    # Block file:// scheme
     if parsed.scheme == "file":
         raise ValueError(
             f"file:// URLs are not allowed: '{url}'. "
             f"Use a file path instead, or set {_UNSAFE_PATHS_ENV}=true to bypass."
         )
 
-    # Only allow http and https
     if parsed.scheme not in ("http", "https"):
         raise ValueError(
             f"URL scheme '{parsed.scheme}' is not allowed. Only http and https are supported."
@@ -208,6 +195,7 @@ def validate_url(url: str) -> str:
     except socket.gaierror as exc:
         raise ValueError(f"Could not resolve hostname: '{parsed.hostname}'") from exc
 
+    resolved_ip = None
     for _family, _, _, _, sockaddr in addrinfos:
         ip_str = str(sockaddr[0])
         if _is_private_or_reserved(ip_str):
@@ -216,5 +204,115 @@ def validate_url(url: str) -> str:
                 f"Access to internal networks is not allowed. "
                 f"Set {_UNSAFE_PATHS_ENV}=true to bypass."
             )
+        if resolved_ip is None:
+            resolved_ip = ip_str
+
+    return parsed.scheme, parsed.hostname, resolved_ip
+
+
+def validate_url(url: str) -> str:
+    """Validate that a URL is safe to fetch.
+
+    Blocks ``file://`` scheme entirely. For ``http``/``https``, resolves
+    DNS and checks that the target IP is not private or reserved (prevents
+    SSRF to internal services and cloud metadata endpoints).
+
+    Args:
+        url: The URL to validate.
+
+    Returns:
+        The validated URL string.
 
+    Raises:
+        ValueError: If the URL uses a blocked scheme or resolves to a
+            private/reserved IP address.
+    """
+    _validate_url_common(url)
     return url
+
+
+@dataclass(frozen=True, slots=True)
+class ValidatedURL:
+    """A validated URL with its pinned resolved IP address.
+
+    The resolved IP is captured during validation and used by
+    :class:`PinnedIPAdapter` to eliminate the TOCTOU window between
+    DNS resolution and TCP connection.
+    """
+
+    url: str
+    resolved_ip: str
+
+
+def validate_url_and_resolve(url: str) -> ValidatedURL:
+    """Validate a URL and return its resolved IP for pinning.
+
+    Same security checks as :func:`validate_url`, but additionally returns
+    the resolved IP address so the caller can pin the connection to it,
+    eliminating the DNS-rebinding TOCTOU window.
+
+    Args:
+        url: The URL to validate.
+
+    Returns:
+        A :class:`ValidatedURL` with the original URL and the resolved IP.
+
+    Raises:
+        ValueError: If the URL is unsafe or unresolvable.
+    """
+    _, _, resolved_ip = _validate_url_common(url)
+    if resolved_ip is None:
+        return ValidatedURL(url=url, resolved_ip="")
+    return ValidatedURL(url=url, resolved_ip=resolved_ip)
+
+
+class PinnedIPAdapter(HTTPAdapter):
+    """urllib3 adapter that pins connections to a pre-resolved IP.
+
+    This prevents DNS rebinding attacks: the IP is validated once, then
+    the connection is made directly to that IP while preserving the
+    original hostname for ``Host`` header and TLS SNI.
+
+    Implementation: overrides ``send()`` to rewrite the request URL,
+    replacing the hostname with the pinned IP and setting the ``Host``
+    header to the original hostname.  ``init_poolmanager`` is overridden
+    to configure ``assert_hostname=False`` so that TLS SNI verification
+    works correctly when the URL contains an IP address.
+    """
+
+    def __init__(self, resolved_ip: str, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._resolved_ip = resolved_ip
+
+    def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:
+        """Configure pool manager with assert_hostname=False for IP-pinned HTTPS."""
+        super().init_poolmanager(*args, **kwargs)  # type: ignore[no-untyped-call]
+        self.poolmanager.assert_hostname = False
+
+    def send(self, request: Any, **kwargs: Any) -> Any:  # type: ignore[override]
+        """Send the request, rewriting the URL to use the pinned IP."""
+        from urllib.parse import urlparse, urlunparse
+
+        parsed = urlparse(request.url)
+        original_host = parsed.hostname
+        if not original_host or not self._resolved_ip:
+            return super().send(request, **kwargs)
+
+        # Rewrite URL: replace hostname with pinned IP
+        pinned_netloc = self._resolved_ip
+        if parsed.port:
+            pinned_netloc = f"{self._resolved_ip}:{parsed.port}"
+        if parsed.username:
+            auth = parsed.username
+            if parsed.password:
+                auth += f":{parsed.password}"
+            pinned_netloc = f"{auth}@{pinned_netloc}"
+
+        pinned_url = urlunparse(parsed._replace(netloc=pinned_netloc))
+        request.url = pinned_url
+
+        # Set Host header to original hostname for virtual hosting
+        if "host" not in {k.lower() for k in request.headers}:
+            request.headers["Host"] = original_host
+
+        return super().send(request, **kwargs)

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

@@ -7,7 +7,10 @@
 
 import requests
 
-from crewai_tools.security.safe_path import validate_url
+from crewai_tools.security.safe_path import (
+    ValidatedURL,
+    validate_url_and_resolve,
+)
 
 
 _REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
@@ -48,16 +51,50 @@ def _strip_cross_origin_credentials(request_kwargs: dict[str, Any]) -> dict[str,
     return sanitized
 
 
+def _build_pinned_session(validated: ValidatedURL) -> requests.Session:
+    """Build a requests.Session that pins TCP connections to the resolved IP.
+
+    This eliminates the DNS-rebinding TOCTOU window: the IP is validated
+    once during ``validate_url_and_resolve``, and the session's adapter
+    connects directly to that IP while preserving the original hostname
+    for ``Host`` header and TLS SNI.
+    """
+    session = requests.Session()
+    if not validated.resolved_ip:
+        return session
+
+    from crewai_tools.security.safe_path import PinnedIPAdapter
+
+    prefix = "https://" if urlparse(validated.url).scheme == "https" else "http://"
+    adapter = PinnedIPAdapter(resolved_ip=validated.resolved_ip)
+    session.mount(prefix, adapter)
+    return session
+
+
 def safe_get(url: str, *, max_redirects: int = 10, **kwargs: Any) -> requests.Response:
-    """GET a URL while validating each redirect target before following it."""
-    current_url = validate_url(url)
+    """GET a URL while validating each redirect target before following it.
+
+    Uses IP pinning to prevent DNS-rebinding TOCTOU attacks: the DNS is
+    resolved once during validation, and the actual connection is made
+    directly to the resolved IP.
+    """
+    validated = validate_url_and_resolve(url)
+    current_url = validated.url
+    current_ip = validated.resolved_ip
     request_kwargs = {**kwargs, "allow_redirects": False}
     timeout = request_kwargs.pop("timeout", 30)
     history: list[requests.Response] = []
     redirects_followed = 0
 
     while True:
-        response = requests.get(current_url, timeout=timeout, **request_kwargs)
+        session = _build_pinned_session(
+            ValidatedURL(url=current_url, resolved_ip=current_ip)
+        )
+        try:
+            response = session.get(current_url, timeout=timeout, **request_kwargs)
+        finally:
+            session.close()
+
         if (
             response.status_code not in _REDIRECT_STATUS_CODES
             or "Location" not in response.headers
@@ -75,14 +112,17 @@ def safe_get(url: str, *, max_redirects: int = 10, **kwargs: Any) -> requests.Re
             return response
 
         try:
-            redirect_url = validate_url(urljoin(response.url, location))
+            redirect_validated = validate_url_and_resolve(
+                urljoin(response.url, location)
+            )
         except ValueError:
             response.close()
             raise
 
-        if not _same_origin(current_url, redirect_url):
+        if not _same_origin(current_url, redirect_validated.url):
             request_kwargs = _strip_cross_origin_credentials(request_kwargs)
 
         history.append(response)
-        current_url = redirect_url
+        current_url = redirect_validated.url
+        current_ip = redirect_validated.resolved_ip
         redirects_followed += 1