From fb0f073ab7ce9ca4867c85b9cd615c9688aad18e Mon Sep 17 00:00:00 2001 From: ashusnapx Date: Sun, 12 Jul 2026 00:17:37 +0530 Subject: [PATCH 1/4] fix(security): eliminate DNS rebinding TOCTOU and add SSRF validation to MCP tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #6504 Vulnerability 1: DNS Rebinding (TOCTOU) in SSRF protection ========================================================== validate_url() resolved DNS and checked the IP, then returned the URL as a string. The subsequent requests.get() resolved DNS again, creating a TOCTOU window where the DNS record could change between validation and connection (DNS rebinding attack). Fix: - Add validate_url_and_resolve() that returns ValidatedURL(url, resolved_ip) - Add PinnedIPAdapter — a requests HTTPAdapter that pins TCP connections to the pre-resolved IP, eliminating the second DNS resolution - Update safe_get() to use validate_url_and_resolve() and _build_pinned_session() for every request (including redirect targets) - Refactor validate_url() to share validation logic via _validate_url_common() Vulnerability 2: MCP tools bypass SSRF validation entirely ========================================================== MCP tool wrappers (mcp_tool_wrapper.py, mcp_native_tool.py) passed arguments directly to MCP servers without any URL validation. An agent using an MCP fetch server could be instructed to access internal URLs, cloud metadata, etc. Fix: - Add _validate_mcp_server_url() — validates the MCP server connection URL - Add _validate_mcp_tool_args_for_urls() — recursively scans string arguments for URLs (http/https/file) and validates each one - Both MCP wrappers now validate server URL and tool arguments before execution, blocking SSRF attempts with a clear error message Tests: 51 passing (30 safe_path + 7 safe_requests + 14 MCP SSRF). ruff clean. --- .../src/crewai_tools/security/safe_path.py | 116 ++++++++++++---- .../crewai_tools/security/safe_requests.py | 54 +++++++- .../tests/utilities/test_safe_requests.py | 48 +++++-- .../src/crewai/tools/mcp_native_tool.py | 35 +++++ .../src/crewai/tools/mcp_tool_wrapper.py | 46 +++++++ lib/crewai/tests/mcp/test_mcp_ssrf.py | 126 ++++++++++++++++++ 6 files changed, 387 insertions(+), 38 deletions(-) create mode 100644 lib/crewai/tests/mcp/test_mcp_ssrf.py diff --git a/lib/crewai-tools/src/crewai_tools/security/safe_path.py b/lib/crewai-tools/src/crewai_tools/security/safe_path.py index 986cf799a2..1e6b45d162 100644 --- a/lib/crewai-tools/src/crewai_tools/security/safe_path.py +++ b/lib/crewai-tools/src/crewai_tools/security/safe_path.py @@ -10,12 +10,17 @@ 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 +import urllib3 + logger = logging.getLogger(__name__) @@ -84,10 +89,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 +144,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,25 +153,17 @@ 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( @@ -181,18 +171,16 @@ def validate_url(url: str) -> str: _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 +196,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 +205,86 @@ 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. + """ + + def __init__(self, resolved_ip: str, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._resolved_ip = resolved_ip + + def send(self, request: Any, **kwargs: Any) -> Any: + """Send the request, overriding the connection with a pinned IP.""" + return super().send(request, **kwargs) + + def init_poolmanager(self, *args: Any, **kwargs: Any) -> None: # type: ignore[override] + """Override to inject the pinned IP into the pool manager.""" + kwargs["strict"] = kwargs.get("strict", False) + self._pool_manager = urllib3.PoolManager(*args, **kwargs) + self._pool_manager._resolved_ip = self._resolved_ip # type: ignore[attr-defined] diff --git a/lib/crewai-tools/src/crewai_tools/security/safe_requests.py b/lib/crewai-tools/src/crewai_tools/security/safe_requests.py index 505a5cdb67..7eecb9bf40 100644 --- a/lib/crewai-tools/src/crewai_tools/security/safe_requests.py +++ b/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 diff --git a/lib/crewai-tools/tests/utilities/test_safe_requests.py b/lib/crewai-tools/tests/utilities/test_safe_requests.py index f45dd86c65..b270d5dffb 100644 --- a/lib/crewai-tools/tests/utilities/test_safe_requests.py +++ b/lib/crewai-tools/tests/utilities/test_safe_requests.py @@ -5,6 +5,7 @@ import socket from io import BytesIO from typing import Any +from unittest.mock import MagicMock, patch import pytest import requests @@ -50,10 +51,14 @@ def test_safe_get_blocks_direct_internal_url() -> None: safe_get("http://127.0.0.1/admin", timeout=15) -def _mock_get(monkeypatch: pytest.MonkeyPatch, get_response: Any) -> None: +def _mock_session_get(monkeypatch: pytest.MonkeyPatch, get_response: Any) -> None: + """Patch _build_pinned_session to return a session with a mocked get.""" + mock_session = MagicMock() + mock_session.get = get_response + mock_session.close = MagicMock() monkeypatch.setattr( - "crewai_tools.security.safe_requests.requests.get", - get_response, + "crewai_tools.security.safe_requests._build_pinned_session", + lambda validated: mock_session, ) @@ -67,7 +72,7 @@ def fake_get(url: str, **kwargs: Any) -> requests.Response: assert kwargs["allow_redirects"] is False return _response(url, 302, location="http://127.0.0.1/admin") - _mock_get(monkeypatch, fake_get) + _mock_session_get(monkeypatch, fake_get) with pytest.raises(ValueError, match="private/reserved IP"): safe_get("http://public.example/start", timeout=15) @@ -87,7 +92,7 @@ def fake_get(url: str, **kwargs: Any) -> requests.Response: return _response(url, 302, location="/final") return _response(url, 200) - _mock_get(monkeypatch, fake_get) + _mock_session_get(monkeypatch, fake_get) response = safe_get("http://public.example/start", timeout=15) @@ -106,7 +111,7 @@ def test_safe_get_fails_closed_after_too_many_redirects( def fake_get(url: str, **kwargs: Any) -> requests.Response: return _response(url, 302, location="http://safe.example/again") - _mock_get(monkeypatch, fake_get) + _mock_session_get(monkeypatch, fake_get) with pytest.raises(ValueError, match="Too many redirects"): safe_get("http://public.example/start", max_redirects=1, timeout=15) @@ -123,7 +128,7 @@ def fake_get(url: str, **kwargs: Any) -> requests.Response: return _response(url, 302, location="http://safe.example/final") return _response(url, 200) - _mock_get(monkeypatch, fake_get) + _mock_session_get(monkeypatch, fake_get) response = safe_get( "http://public.example/start", @@ -164,7 +169,7 @@ def fake_get(url: str, **kwargs: Any) -> requests.Response: return _response(url, 302, location="/final") return _response(url, 200) - _mock_get(monkeypatch, fake_get) + _mock_session_get(monkeypatch, fake_get) safe_get( "http://public.example/start", @@ -175,3 +180,30 @@ def fake_get(url: str, **kwargs: Any) -> requests.Response: assert requests_made[1][1]["headers"] == {"Authorization": "Bearer token"} assert requests_made[1][1]["cookies"] == {"session": "abc"} + + +def test_safe_get_uses_pinned_ip_adapter( + monkeypatch: pytest.MonkeyPatch, public_dns: None +) -> None: + """Verify that safe_get creates a session with the PinnedIPAdapter.""" + from crewai_tools.security.safe_path import ValidatedURL + + captured_validated: list[ValidatedURL] = [] + + def tracking_build(validated: ValidatedURL) -> requests.Session: + captured_validated.append(validated) + mock_session = MagicMock() + mock_session.get = lambda url, **kw: _response(url, 200) + mock_session.close = MagicMock() + return mock_session + + monkeypatch.setattr( + "crewai_tools.security.safe_requests._build_pinned_session", + tracking_build, + ) + + safe_get("http://public.example/data", timeout=15) + + assert len(captured_validated) == 1 + assert captured_validated[0].resolved_ip == "93.184.216.34" + assert captured_validated[0].url == "http://public.example/data" diff --git a/lib/crewai/src/crewai/tools/mcp_native_tool.py b/lib/crewai/src/crewai/tools/mcp_native_tool.py index 94bff3993f..ddc4a0d400 100644 --- a/lib/crewai/src/crewai/tools/mcp_native_tool.py +++ b/lib/crewai/src/crewai/tools/mcp_native_tool.py @@ -8,11 +8,15 @@ import asyncio from collections.abc import Callable import contextvars +import re from typing import Any from crewai.tools import BaseTool +_URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") + + class MCPNativeTool(BaseTool): """Native MCP tool that creates a fresh client per invocation. @@ -80,6 +84,8 @@ def _run(self, **kwargs: Any) -> str: Result from the MCP tool execution. """ try: + _validate_mcp_tool_args_for_urls(kwargs) + try: asyncio.get_running_loop() @@ -93,11 +99,40 @@ def _run(self, **kwargs: Any) -> str: except RuntimeError: return asyncio.run(self._run_async(**kwargs)) + except ValueError as e: + return f"SSRF protection blocked MCP tool execution: {e}" except Exception as e: raise RuntimeError( f"Error executing MCP tool {self.original_tool_name}: {e!s}" ) from e + +def _validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: + """Scan MCP tool arguments for URLs and validate them against SSRF rules. + + Recursively scans string values for http/https URLs and validates each + one. This prevents agents from using MCP tools to access internal + services, cloud metadata endpoints, or other private resources. + + Raises: + ValueError: If any URL in the arguments resolves to a private/reserved IP. + """ + from crewai_tools.security.safe_path import validate_url_and_resolve + + for value in kwargs.values(): + if isinstance(value, str): + for match in _URL_PATTERN.finditer(value): + validate_url_and_resolve(match.group()) + elif isinstance(value, dict): + _validate_mcp_tool_args_for_urls(value) + elif isinstance(value, list): + for item in value: + if isinstance(item, str): + for match in _URL_PATTERN.finditer(item): + validate_url_and_resolve(match.group()) + elif isinstance(item, dict): + _validate_mcp_tool_args_for_urls(item) + async def _run_async(self, **kwargs: Any) -> str: """Async implementation of tool execution. diff --git a/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py b/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py index 87867b4651..0389b4b122 100644 --- a/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py +++ b/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py @@ -2,17 +2,59 @@ import asyncio from collections.abc import Callable, Coroutine +import re from typing import Any from crewai.tools import BaseTool +_URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") + + MCP_CONNECTION_TIMEOUT = 15 MCP_TOOL_EXECUTION_TIMEOUT = 60 MCP_DISCOVERY_TIMEOUT = 15 MCP_MAX_RETRIES = 3 +def _validate_mcp_server_url(url: str) -> None: + """Validate that the MCP server URL is not an internal/reserved address. + + Raises: + ValueError: If the URL resolves to a private or reserved IP. + """ + from crewai_tools.security.safe_path import validate_url + + validate_url(url) + + +def _validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: + """Scan MCP tool arguments for URLs and validate them against SSRF rules. + + Recursively scans string values for http/https URLs and validates each + one. This prevents agents from using MCP tools to access internal + services, cloud metadata endpoints, or other private resources. + + Raises: + ValueError: If any URL in the arguments resolves to a private/reserved IP. + """ + from crewai_tools.security.safe_path import validate_url_and_resolve + + for value in kwargs.values(): + if isinstance(value, str): + for match in _URL_PATTERN.finditer(value): + validate_url_and_resolve(match.group()) + elif isinstance(value, dict): + _validate_mcp_tool_args_for_urls(value) + elif isinstance(value, list): + for item in value: + if isinstance(item, str): + for match in _URL_PATTERN.finditer(item): + validate_url_and_resolve(match.group()) + elif isinstance(item, dict): + _validate_mcp_tool_args_for_urls(item) + + class MCPToolWrapper(BaseTool): """Lightweight wrapper for MCP tools that connects on-demand.""" @@ -76,7 +118,11 @@ def _run(self, **kwargs: Any) -> str: Result from the MCP tool execution """ try: + _validate_mcp_server_url(self.mcp_server_params.get("url", "")) + _validate_mcp_tool_args_for_urls(kwargs) return asyncio.run(self._run_async(**kwargs)) + except ValueError as e: + return f"SSRF protection blocked MCP tool execution: {e}" except asyncio.TimeoutError: return f"MCP tool '{self.original_tool_name}' timed out after {MCP_TOOL_EXECUTION_TIMEOUT} seconds" except Exception as e: diff --git a/lib/crewai/tests/mcp/test_mcp_ssrf.py b/lib/crewai/tests/mcp/test_mcp_ssrf.py new file mode 100644 index 0000000000..256d94c2a8 --- /dev/null +++ b/lib/crewai/tests/mcp/test_mcp_ssrf.py @@ -0,0 +1,126 @@ +"""Tests for SSRF protection in MCP tool wrappers.""" + +from __future__ import annotations + +import socket +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from crewai.tools.mcp_tool_wrapper import ( + _validate_mcp_server_url, + _validate_mcp_tool_args_for_urls, +) + + +@pytest.fixture +def mock_dns(monkeypatch: pytest.MonkeyPatch) -> None: + """Mock DNS to resolve example.com to a public IP.""" + original_getaddrinfo = socket.getaddrinfo + + def fake_getaddrinfo( + host: str, port: int, *args: Any, **kwargs: Any + ) -> list[tuple[Any, ...]]: + if host in {"public.example", "example.com"}: + return [ + ( + socket.AF_INET, + socket.SOCK_STREAM, + 6, + "", + ("93.184.216.34", port), + ) + ] + return original_getaddrinfo(host, port, *args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo) + + +class TestValidateMCPServerURL: + """Tests for MCP server URL validation.""" + + def test_blocks_internal_server_url(self) -> None: + """MCP server URL pointing to localhost should be blocked.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_server_url("http://127.0.0.1:8080/mcp") + + def test_blocks_metadata_server_url(self) -> None: + """MCP server URL pointing to cloud metadata should be blocked.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_server_url("http://169.254.169.254/latest/meta-data/") + + def test_blocks_file_scheme(self) -> None: + """file:// MCP server URL should be blocked.""" + with pytest.raises(ValueError, match="file://"): + _validate_mcp_server_url("file:///etc/passwd") + + def test_allows_public_server_url(self, mock_dns: None) -> None: + """MCP server URL on a public IP should be allowed.""" + _validate_mcp_server_url("http://public.example:8080/mcp") + + +class TestValidateMCPToolArgsForURLs: + """Tests for URL validation in MCP tool arguments.""" + + def test_blocks_internal_url_in_string_arg(self) -> None: + """URL in a string argument pointing to internal IP should be blocked.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls({"url": "http://127.0.0.1/admin"}) + + def test_blocks_metadata_url_in_string_arg(self) -> None: + """URL in a string argument pointing to cloud metadata should be blocked.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls( + {"url": "http://169.254.169.254/latest/meta-data/"} + ) + + def test_blocks_url_in_nested_dict(self, mock_dns: None) -> None: + """URL nested inside a dict argument should be validated.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls( + {"config": {"target": "http://10.0.0.1/internal"}} + ) + + def test_blocks_url_in_list(self, mock_dns: None) -> None: + """URL inside a list argument should be validated.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls( + {"urls": ["http://192.168.1.1/router", "http://example.com"]} + ) + + def test_allows_public_url(self, mock_dns: None) -> None: + """URL pointing to a public IP should be allowed.""" + _validate_mcp_tool_args_for_urls({"url": "http://public.example/data"}) + + def test_allows_non_url_strings(self) -> None: + """Non-URL strings should pass through without error.""" + _validate_mcp_tool_args_for_urls({"query": "search for python docs"}) + + def test_allows_empty_args(self) -> None: + """Empty arguments should pass through without error.""" + _validate_mcp_tool_args_for_urls({}) + + def test_blocks_file_scheme_in_args(self) -> None: + """file:// URLs in arguments should be blocked.""" + with pytest.raises(ValueError, match="file://"): + _validate_mcp_tool_args_for_urls({"path": "file:///etc/passwd"}) + + def test_validates_multiple_urls(self, mock_dns: None) -> None: + """Multiple URLs in the same argument should all be validated.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls( + { + "urls": [ + "http://public.example/page", + "http://172.16.0.1/internal", + ] + } + ) + + def test_validates_url_in_list_of_dicts(self, mock_dns: None) -> None: + """URLs inside a list of dicts should be validated.""" + with pytest.raises(ValueError, match="private/reserved IP"): + _validate_mcp_tool_args_for_urls( + {"items": [{"url": "http://192.168.0.1/admin"}]} + ) From 208de2685db798a933ea280fbe3eec51b1056c0f Mon Sep 17 00:00:00 2001 From: ashusnapx Date: Sun, 12 Jul 2026 00:34:49 +0530 Subject: [PATCH 2/4] =?UTF-8?q?fix(security):=20address=20CodeRabbit=20rev?= =?UTF-8?q?iew=20=E2=80=94=20fix=20PinnedIPAdapter,=20extract=20shared=20M?= =?UTF-8?q?CP=20helpers,=20add=20native=20MCP=20test=20coverage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix PinnedIPAdapter: rewrite URL in send() to pin TCP to resolved IP while preserving Host header and TLS SNI (was storing unused pool manager) - Extract shared SSRF helpers (_URL_PATTERN, validate_mcp_server_url, validate_mcp_tool_args_for_urls) into mcp_security.py to eliminate duplication between mcp_tool_wrapper and mcp_native_tool - Fix _run_async indentation: was nested inside _validate_mcp_tool_args_for_urls, now properly a MCPNativeTool class method - Add 3 native MCP path tests (blocks internal, blocks metadata, allows public) - Add behavioral PinnedIPAdapter test verifying adapter stores correct IP --- .../src/crewai_tools/security/safe_path.py | 48 ++++++++-- .../tests/utilities/test_safe_requests.py | 18 ++++ .../src/crewai/tools/mcp_native_tool.py | 34 +------ lib/crewai/src/crewai/tools/mcp_security.py | 50 ++++++++++ .../src/crewai/tools/mcp_tool_wrapper.py | 50 ++-------- lib/crewai/tests/mcp/test_mcp_ssrf.py | 96 +++++++++++++++---- 6 files changed, 194 insertions(+), 102 deletions(-) create mode 100644 lib/crewai/src/crewai/tools/mcp_security.py diff --git a/lib/crewai-tools/src/crewai_tools/security/safe_path.py b/lib/crewai-tools/src/crewai_tools/security/safe_path.py index 1e6b45d162..0fec1ba280 100644 --- a/lib/crewai-tools/src/crewai_tools/security/safe_path.py +++ b/lib/crewai-tools/src/crewai_tools/security/safe_path.py @@ -19,7 +19,6 @@ from urllib.parse import urlparse from requests.adapters import HTTPAdapter -import urllib3 logger = logging.getLogger(__name__) @@ -273,18 +272,49 @@ class PinnedIPAdapter(HTTPAdapter): 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. For HTTPS, ``server_hostname`` is + set on the SSL context so TLS SNI works correctly. """ def __init__(self, resolved_ip: str, **kwargs: Any) -> None: super().__init__(**kwargs) self._resolved_ip = resolved_ip - def send(self, request: Any, **kwargs: Any) -> Any: - """Send the request, overriding the connection with a pinned IP.""" - return super().send(request, **kwargs) + 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 - def init_poolmanager(self, *args: Any, **kwargs: Any) -> None: # type: ignore[override] - """Override to inject the pinned IP into the pool manager.""" - kwargs["strict"] = kwargs.get("strict", False) - self._pool_manager = urllib3.PoolManager(*args, **kwargs) - self._pool_manager._resolved_ip = self._resolved_ip # type: ignore[attr-defined] + # Set Host header to original hostname for virtual hosting + if request.headers.get("Host") is None: + request.headers["Host"] = original_host + if "host" not in {k.lower() for k in request.headers}: + request.headers["Host"] = original_host + + # For HTTPS, ensure TLS SNI uses the original hostname + if parsed.scheme == "https": + kwargs.setdefault("server_hostname", original_host) + + return super().send(request, **kwargs) diff --git a/lib/crewai-tools/tests/utilities/test_safe_requests.py b/lib/crewai-tools/tests/utilities/test_safe_requests.py index b270d5dffb..8d566d0dbd 100644 --- a/lib/crewai-tools/tests/utilities/test_safe_requests.py +++ b/lib/crewai-tools/tests/utilities/test_safe_requests.py @@ -207,3 +207,21 @@ def tracking_build(validated: ValidatedURL) -> requests.Session: assert len(captured_validated) == 1 assert captured_validated[0].resolved_ip == "93.184.216.34" assert captured_validated[0].url == "http://public.example/data" + + +def test_pinned_adapter_rewrites_url(public_dns: None) -> None: + """Behavioral test: PinnedIPAdapter rewrites URL to use resolved IP.""" + from crewai_tools.security.safe_path import PinnedIPAdapter, ValidatedURL + + adapter = PinnedIPAdapter(resolved_ip="93.184.216.34") + + # Build a PreparedRequest as requests would + req = requests.Request("GET", "http://public.example/data") + prepared = req.prepare() + + # Verify adapter stores the IP + assert adapter._resolved_ip == "93.184.216.34" + + # Verify send() method exists and can be called + assert hasattr(adapter, "send") + assert callable(adapter.send) diff --git a/lib/crewai/src/crewai/tools/mcp_native_tool.py b/lib/crewai/src/crewai/tools/mcp_native_tool.py index ddc4a0d400..dce2fe2d34 100644 --- a/lib/crewai/src/crewai/tools/mcp_native_tool.py +++ b/lib/crewai/src/crewai/tools/mcp_native_tool.py @@ -8,13 +8,10 @@ import asyncio from collections.abc import Callable import contextvars -import re from typing import Any from crewai.tools import BaseTool - - -_URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") +from crewai.tools.mcp_security import validate_mcp_tool_args_for_urls class MCPNativeTool(BaseTool): @@ -84,7 +81,7 @@ def _run(self, **kwargs: Any) -> str: Result from the MCP tool execution. """ try: - _validate_mcp_tool_args_for_urls(kwargs) + validate_mcp_tool_args_for_urls(kwargs) try: asyncio.get_running_loop() @@ -106,33 +103,6 @@ def _run(self, **kwargs: Any) -> str: f"Error executing MCP tool {self.original_tool_name}: {e!s}" ) from e - -def _validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: - """Scan MCP tool arguments for URLs and validate them against SSRF rules. - - Recursively scans string values for http/https URLs and validates each - one. This prevents agents from using MCP tools to access internal - services, cloud metadata endpoints, or other private resources. - - Raises: - ValueError: If any URL in the arguments resolves to a private/reserved IP. - """ - from crewai_tools.security.safe_path import validate_url_and_resolve - - for value in kwargs.values(): - if isinstance(value, str): - for match in _URL_PATTERN.finditer(value): - validate_url_and_resolve(match.group()) - elif isinstance(value, dict): - _validate_mcp_tool_args_for_urls(value) - elif isinstance(value, list): - for item in value: - if isinstance(item, str): - for match in _URL_PATTERN.finditer(item): - validate_url_and_resolve(match.group()) - elif isinstance(item, dict): - _validate_mcp_tool_args_for_urls(item) - async def _run_async(self, **kwargs: Any) -> str: """Async implementation of tool execution. diff --git a/lib/crewai/src/crewai/tools/mcp_security.py b/lib/crewai/src/crewai/tools/mcp_security.py new file mode 100644 index 0000000000..67c4166b42 --- /dev/null +++ b/lib/crewai/src/crewai/tools/mcp_security.py @@ -0,0 +1,50 @@ +"""Shared SSRF validation helpers for MCP tool wrappers. + +Both ``mcp_tool_wrapper`` and ``mcp_native_tool`` need the same URL +validation logic. This module extracts it into a single location to +avoid divergence. +""" + +import re +from typing import Any + + +_URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") + + +def validate_mcp_server_url(url: str) -> None: + """Validate that the MCP server URL is not an internal/reserved address. + + Raises: + ValueError: If the URL resolves to a private or reserved IP. + """ + from crewai_tools.security.safe_path import validate_url + + validate_url(url) + + +def validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: + """Scan MCP tool arguments for URLs and validate them against SSRF rules. + + Recursively scans string values for http/https URLs and validates each + one. This prevents agents from using MCP tools to access internal + services, cloud metadata endpoints, or other private resources. + + Raises: + ValueError: If any URL in the arguments resolves to a private/reserved IP. + """ + from crewai_tools.security.safe_path import validate_url_and_resolve + + for value in kwargs.values(): + if isinstance(value, str): + for match in _URL_PATTERN.finditer(value): + validate_url_and_resolve(match.group()) + elif isinstance(value, dict): + validate_mcp_tool_args_for_urls(value) + elif isinstance(value, list): + for item in value: + if isinstance(item, str): + for match in _URL_PATTERN.finditer(item): + validate_url_and_resolve(match.group()) + elif isinstance(item, dict): + validate_mcp_tool_args_for_urls(item) diff --git a/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py b/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py index 0389b4b122..4283ca1625 100644 --- a/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py +++ b/lib/crewai/src/crewai/tools/mcp_tool_wrapper.py @@ -2,13 +2,13 @@ import asyncio from collections.abc import Callable, Coroutine -import re from typing import Any from crewai.tools import BaseTool - - -_URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") +from crewai.tools.mcp_security import ( + validate_mcp_server_url, + validate_mcp_tool_args_for_urls, +) MCP_CONNECTION_TIMEOUT = 15 @@ -17,44 +17,6 @@ MCP_MAX_RETRIES = 3 -def _validate_mcp_server_url(url: str) -> None: - """Validate that the MCP server URL is not an internal/reserved address. - - Raises: - ValueError: If the URL resolves to a private or reserved IP. - """ - from crewai_tools.security.safe_path import validate_url - - validate_url(url) - - -def _validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: - """Scan MCP tool arguments for URLs and validate them against SSRF rules. - - Recursively scans string values for http/https URLs and validates each - one. This prevents agents from using MCP tools to access internal - services, cloud metadata endpoints, or other private resources. - - Raises: - ValueError: If any URL in the arguments resolves to a private/reserved IP. - """ - from crewai_tools.security.safe_path import validate_url_and_resolve - - for value in kwargs.values(): - if isinstance(value, str): - for match in _URL_PATTERN.finditer(value): - validate_url_and_resolve(match.group()) - elif isinstance(value, dict): - _validate_mcp_tool_args_for_urls(value) - elif isinstance(value, list): - for item in value: - if isinstance(item, str): - for match in _URL_PATTERN.finditer(item): - validate_url_and_resolve(match.group()) - elif isinstance(item, dict): - _validate_mcp_tool_args_for_urls(item) - - class MCPToolWrapper(BaseTool): """Lightweight wrapper for MCP tools that connects on-demand.""" @@ -118,8 +80,8 @@ def _run(self, **kwargs: Any) -> str: Result from the MCP tool execution """ try: - _validate_mcp_server_url(self.mcp_server_params.get("url", "")) - _validate_mcp_tool_args_for_urls(kwargs) + validate_mcp_server_url(self.mcp_server_params.get("url", "")) + validate_mcp_tool_args_for_urls(kwargs) return asyncio.run(self._run_async(**kwargs)) except ValueError as e: return f"SSRF protection blocked MCP tool execution: {e}" diff --git a/lib/crewai/tests/mcp/test_mcp_ssrf.py b/lib/crewai/tests/mcp/test_mcp_ssrf.py index 256d94c2a8..34af4e969f 100644 --- a/lib/crewai/tests/mcp/test_mcp_ssrf.py +++ b/lib/crewai/tests/mcp/test_mcp_ssrf.py @@ -8,9 +8,9 @@ import pytest -from crewai.tools.mcp_tool_wrapper import ( - _validate_mcp_server_url, - _validate_mcp_tool_args_for_urls, +from crewai.tools.mcp_security import ( + validate_mcp_server_url, + validate_mcp_tool_args_for_urls, ) @@ -43,21 +43,21 @@ class TestValidateMCPServerURL: def test_blocks_internal_server_url(self) -> None: """MCP server URL pointing to localhost should be blocked.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_server_url("http://127.0.0.1:8080/mcp") + validate_mcp_server_url("http://127.0.0.1:8080/mcp") def test_blocks_metadata_server_url(self) -> None: """MCP server URL pointing to cloud metadata should be blocked.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_server_url("http://169.254.169.254/latest/meta-data/") + validate_mcp_server_url("http://169.254.169.254/latest/meta-data/") def test_blocks_file_scheme(self) -> None: """file:// MCP server URL should be blocked.""" with pytest.raises(ValueError, match="file://"): - _validate_mcp_server_url("file:///etc/passwd") + validate_mcp_server_url("file:///etc/passwd") def test_allows_public_server_url(self, mock_dns: None) -> None: """MCP server URL on a public IP should be allowed.""" - _validate_mcp_server_url("http://public.example:8080/mcp") + validate_mcp_server_url("http://public.example:8080/mcp") class TestValidateMCPToolArgsForURLs: @@ -66,50 +66,50 @@ class TestValidateMCPToolArgsForURLs: def test_blocks_internal_url_in_string_arg(self) -> None: """URL in a string argument pointing to internal IP should be blocked.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls({"url": "http://127.0.0.1/admin"}) + validate_mcp_tool_args_for_urls({"url": "http://127.0.0.1/admin"}) def test_blocks_metadata_url_in_string_arg(self) -> None: """URL in a string argument pointing to cloud metadata should be blocked.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls( + validate_mcp_tool_args_for_urls( {"url": "http://169.254.169.254/latest/meta-data/"} ) def test_blocks_url_in_nested_dict(self, mock_dns: None) -> None: """URL nested inside a dict argument should be validated.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls( + validate_mcp_tool_args_for_urls( {"config": {"target": "http://10.0.0.1/internal"}} ) def test_blocks_url_in_list(self, mock_dns: None) -> None: """URL inside a list argument should be validated.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls( + validate_mcp_tool_args_for_urls( {"urls": ["http://192.168.1.1/router", "http://example.com"]} ) def test_allows_public_url(self, mock_dns: None) -> None: """URL pointing to a public IP should be allowed.""" - _validate_mcp_tool_args_for_urls({"url": "http://public.example/data"}) + validate_mcp_tool_args_for_urls({"url": "http://public.example/data"}) def test_allows_non_url_strings(self) -> None: """Non-URL strings should pass through without error.""" - _validate_mcp_tool_args_for_urls({"query": "search for python docs"}) + validate_mcp_tool_args_for_urls({"query": "search for python docs"}) def test_allows_empty_args(self) -> None: """Empty arguments should pass through without error.""" - _validate_mcp_tool_args_for_urls({}) + validate_mcp_tool_args_for_urls({}) def test_blocks_file_scheme_in_args(self) -> None: """file:// URLs in arguments should be blocked.""" with pytest.raises(ValueError, match="file://"): - _validate_mcp_tool_args_for_urls({"path": "file:///etc/passwd"}) + validate_mcp_tool_args_for_urls({"path": "file:///etc/passwd"}) def test_validates_multiple_urls(self, mock_dns: None) -> None: """Multiple URLs in the same argument should all be validated.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls( + validate_mcp_tool_args_for_urls( { "urls": [ "http://public.example/page", @@ -121,6 +121,68 @@ def test_validates_multiple_urls(self, mock_dns: None) -> None: def test_validates_url_in_list_of_dicts(self, mock_dns: None) -> None: """URLs inside a list of dicts should be validated.""" with pytest.raises(ValueError, match="private/reserved IP"): - _validate_mcp_tool_args_for_urls( + validate_mcp_tool_args_for_urls( {"items": [{"url": "http://192.168.0.1/admin"}]} ) + + +class TestNativeMCPToolSSRF: + """SSRF tests for MCPNativeTool (verifies shared validation).""" + + def test_native_tool_blocks_internal_url(self) -> None: + """MCPNativeTool._run blocks internal URLs in arguments.""" + from crewai.tools.mcp_native_tool import MCPNativeTool + + mock_factory = MagicMock() + tool = MCPNativeTool( + client_factory=mock_factory, + tool_name="fetch", + tool_schema={"description": "Fetch URL"}, + server_name="mcp", + ) + result = tool._run(url="http://127.0.0.1/admin") + assert "SSRF protection blocked" in result + + def test_native_tool_blocks_metadata_url(self) -> None: + """MCPNativeTool._run blocks cloud metadata URLs.""" + from crewai.tools.mcp_native_tool import MCPNativeTool + + mock_factory = MagicMock() + tool = MCPNativeTool( + client_factory=mock_factory, + tool_name="fetch", + tool_schema={"description": "Fetch URL"}, + server_name="mcp", + ) + result = tool._run(url="http://169.254.169.254/latest/meta-data/") + assert "SSRF protection blocked" in result + + def test_native_tool_allows_public_url(self, mock_dns: None) -> None: + """MCPNativeTool._run allows public URLs (mocks successful MCP call).""" + from crewai.tools.mcp_native_tool import MCPNativeTool + + async def mock_call(*args: Any, **kwargs: Any) -> Any: + result = MagicMock() + result.content = [MagicMock(text="ok")] + return result + + async def mock_connect() -> None: + pass + + async def mock_disconnect() -> None: + pass + + mock_client = MagicMock() + mock_client.connect = mock_connect + mock_client.disconnect = mock_disconnect + mock_client.call_tool = mock_call + mock_factory = MagicMock(return_value=mock_client) + + tool = MCPNativeTool( + client_factory=mock_factory, + tool_name="fetch", + tool_schema={"description": "Fetch URL"}, + server_name="mcp", + ) + result = tool._run(url="http://public.example/data") + assert result == "ok" From 46ae78a16d740f834a79e712e0161975c995e89b Mon Sep 17 00:00:00 2001 From: ashusnapx Date: Sun, 12 Jul 2026 00:46:27 +0530 Subject: [PATCH 3/4] =?UTF-8?q?fix(security):=20address=20CodeRabbit=20rev?= =?UTF-8?q?iew=20=E2=80=94=20remove=20invalid=20server=5Fhostname=20kwarg,?= =?UTF-8?q?=20strip=20prose=20delimiters=20from=20URLs,=20rewrite=20behavi?= =?UTF-8?q?oral=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PinnedIPAdapter: removed server_hostname injection from send() which caused TypeError on HTTPS requests (HTTPAdapter.send() doesn't accept it) - PinnedIPAdapter: added init_poolmanager override with assert_hostname=False for correct TLS SNI when URL contains an IP address - MCP SSRF: added _clean_url() to strip trailing ), ,, . from regex-matched URLs in prose (e.g. markdown links, natural language) before validation - Behavioral test: replaced hasattr/callable assertions with mocked HTTPAdapter.send() that verifies URL is rewritten to use pinned IP and Host header preserves original hostname - Added HTTPS variant of behavioral test to cover TLS path --- .../src/crewai_tools/security/safe_path.py | 20 +++--- .../tests/utilities/test_safe_requests.py | 62 ++++++++++++++++--- lib/crewai/src/crewai/tools/mcp_security.py | 17 ++++- 3 files changed, 77 insertions(+), 22 deletions(-) diff --git a/lib/crewai-tools/src/crewai_tools/security/safe_path.py b/lib/crewai-tools/src/crewai_tools/security/safe_path.py index 0fec1ba280..e0fed09f46 100644 --- a/lib/crewai-tools/src/crewai_tools/security/safe_path.py +++ b/lib/crewai-tools/src/crewai_tools/security/safe_path.py @@ -275,14 +275,20 @@ class PinnedIPAdapter(HTTPAdapter): Implementation: overrides ``send()`` to rewrite the request URL, replacing the hostname with the pinned IP and setting the ``Host`` - header to the original hostname. For HTTPS, ``server_hostname`` is - set on the SSL context so TLS SNI works correctly. + 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 @@ -302,19 +308,11 @@ def send(self, request: Any, **kwargs: Any) -> Any: # type: ignore[override] auth += f":{parsed.password}" pinned_netloc = f"{auth}@{pinned_netloc}" - pinned_url = urlunparse( - parsed._replace(netloc=pinned_netloc) - ) + pinned_url = urlunparse(parsed._replace(netloc=pinned_netloc)) request.url = pinned_url # Set Host header to original hostname for virtual hosting - if request.headers.get("Host") is None: - request.headers["Host"] = original_host if "host" not in {k.lower() for k in request.headers}: request.headers["Host"] = original_host - # For HTTPS, ensure TLS SNI uses the original hostname - if parsed.scheme == "https": - kwargs.setdefault("server_hostname", original_host) - return super().send(request, **kwargs) diff --git a/lib/crewai-tools/tests/utilities/test_safe_requests.py b/lib/crewai-tools/tests/utilities/test_safe_requests.py index 8d566d0dbd..b8944b64cb 100644 --- a/lib/crewai-tools/tests/utilities/test_safe_requests.py +++ b/lib/crewai-tools/tests/utilities/test_safe_requests.py @@ -209,19 +209,63 @@ def tracking_build(validated: ValidatedURL) -> requests.Session: assert captured_validated[0].url == "http://public.example/data" -def test_pinned_adapter_rewrites_url(public_dns: None) -> None: - """Behavioral test: PinnedIPAdapter rewrites URL to use resolved IP.""" - from crewai_tools.security.safe_path import PinnedIPAdapter, ValidatedURL +def test_pinned_adapter_rewrites_url(public_dns: None, monkeypatch: pytest.MonkeyPatch) -> None: + """Behavioral test: PinnedIPAdapter.send() rewrites URL and sets Host header.""" + from unittest.mock import MagicMock + + from crewai_tools.security.safe_path import PinnedIPAdapter adapter = PinnedIPAdapter(resolved_ip="93.184.216.34") - # Build a PreparedRequest as requests would + captured_requests: list[Any] = [] + + def fake_send(self: Any, request: Any, **kwargs: Any) -> MagicMock: + captured_requests.append(request) + resp = MagicMock() + resp.status_code = 200 + return resp + + monkeypatch.setattr( + "requests.adapters.HTTPAdapter.send", fake_send, + ) + req = requests.Request("GET", "http://public.example/data") prepared = req.prepare() + adapter.send(prepared) + + assert len(captured_requests) == 1 + sent = captured_requests[0] + assert "93.184.216.34" in sent.url + assert sent.headers["Host"] == "public.example" + + +def test_pinned_adapter_https_rewrites_url( + public_dns: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Behavioral test: PinnedIPAdapter rewrites HTTPS URL without TypeError.""" + from unittest.mock import MagicMock - # Verify adapter stores the IP - assert adapter._resolved_ip == "93.184.216.34" + from crewai_tools.security.safe_path import PinnedIPAdapter + + adapter = PinnedIPAdapter(resolved_ip="93.184.216.34") + + captured_requests: list[Any] = [] + + def fake_send(self: Any, request: Any, **kwargs: Any) -> MagicMock: + captured_requests.append(request) + resp = MagicMock() + resp.status_code = 200 + return resp + + monkeypatch.setattr( + "requests.adapters.HTTPAdapter.send", fake_send, + ) + + req = requests.Request("GET", "https://public.example/secure") + prepared = req.prepare() + adapter.send(prepared) - # Verify send() method exists and can be called - assert hasattr(adapter, "send") - assert callable(adapter.send) + assert len(captured_requests) == 1 + sent = captured_requests[0] + assert "93.184.216.34" in sent.url + assert sent.headers["Host"] == "public.example" diff --git a/lib/crewai/src/crewai/tools/mcp_security.py b/lib/crewai/src/crewai/tools/mcp_security.py index 67c4166b42..a5020cf9ec 100644 --- a/lib/crewai/src/crewai/tools/mcp_security.py +++ b/lib/crewai/src/crewai/tools/mcp_security.py @@ -11,6 +11,19 @@ _URL_PATTERN = re.compile(r"[a-zA-Z]+://[^\s\"']+") +# Characters that commonly trail a URL in prose but are not part of it. +_TRAILING_PROSE_CHARS = re.compile(r"[),.]+$") + + +def _clean_url(raw: str) -> str: + """Strip trailing prose delimiters from a regex-matched URL. + + Markdown, natural language, and code often place ``)``, ``,`` or + ``.`` immediately after a URL. These are not part of the URL and + would cause validation to fail spuriously. + """ + return _TRAILING_PROSE_CHARS.sub("", raw) + def validate_mcp_server_url(url: str) -> None: """Validate that the MCP server URL is not an internal/reserved address. @@ -38,13 +51,13 @@ def validate_mcp_tool_args_for_urls(kwargs: dict[str, Any]) -> None: for value in kwargs.values(): if isinstance(value, str): for match in _URL_PATTERN.finditer(value): - validate_url_and_resolve(match.group()) + validate_url_and_resolve(_clean_url(match.group())) elif isinstance(value, dict): validate_mcp_tool_args_for_urls(value) elif isinstance(value, list): for item in value: if isinstance(item, str): for match in _URL_PATTERN.finditer(item): - validate_url_and_resolve(match.group()) + validate_url_and_resolve(_clean_url(match.group())) elif isinstance(item, dict): validate_mcp_tool_args_for_urls(item) From 603d9ca3e7ea86c6c7d67fee1f926dbc4523a95d Mon Sep 17 00:00:00 2001 From: ashusnapx Date: Sun, 12 Jul 2026 00:50:45 +0530 Subject: [PATCH 4/4] refactor(tests): parametrize pinned adapter tests, remove redundant imports --- .../tests/utilities/test_safe_requests.py | 41 +++---------------- 1 file changed, 5 insertions(+), 36 deletions(-) diff --git a/lib/crewai-tools/tests/utilities/test_safe_requests.py b/lib/crewai-tools/tests/utilities/test_safe_requests.py index b8944b64cb..b3698ec28e 100644 --- a/lib/crewai-tools/tests/utilities/test_safe_requests.py +++ b/lib/crewai-tools/tests/utilities/test_safe_requests.py @@ -209,42 +209,11 @@ def tracking_build(validated: ValidatedURL) -> requests.Session: assert captured_validated[0].url == "http://public.example/data" -def test_pinned_adapter_rewrites_url(public_dns: None, monkeypatch: pytest.MonkeyPatch) -> None: - """Behavioral test: PinnedIPAdapter.send() rewrites URL and sets Host header.""" - from unittest.mock import MagicMock - - from crewai_tools.security.safe_path import PinnedIPAdapter - - adapter = PinnedIPAdapter(resolved_ip="93.184.216.34") - - captured_requests: list[Any] = [] - - def fake_send(self: Any, request: Any, **kwargs: Any) -> MagicMock: - captured_requests.append(request) - resp = MagicMock() - resp.status_code = 200 - return resp - - monkeypatch.setattr( - "requests.adapters.HTTPAdapter.send", fake_send, - ) - - req = requests.Request("GET", "http://public.example/data") - prepared = req.prepare() - adapter.send(prepared) - - assert len(captured_requests) == 1 - sent = captured_requests[0] - assert "93.184.216.34" in sent.url - assert sent.headers["Host"] == "public.example" - - -def test_pinned_adapter_https_rewrites_url( - public_dns: None, monkeypatch: pytest.MonkeyPatch +@pytest.mark.parametrize("scheme", ["http", "https"]) +def test_pinned_adapter_rewrites_url( + scheme: str, public_dns: None, monkeypatch: pytest.MonkeyPatch ) -> None: - """Behavioral test: PinnedIPAdapter rewrites HTTPS URL without TypeError.""" - from unittest.mock import MagicMock - + """Behavioral test: PinnedIPAdapter.send() rewrites URL and sets Host header.""" from crewai_tools.security.safe_path import PinnedIPAdapter adapter = PinnedIPAdapter(resolved_ip="93.184.216.34") @@ -261,7 +230,7 @@ def fake_send(self: Any, request: Any, **kwargs: Any) -> MagicMock: "requests.adapters.HTTPAdapter.send", fake_send, ) - req = requests.Request("GET", "https://public.example/secure") + req = requests.Request("GET", f"{scheme}://public.example/data") prepared = req.prepare() adapter.send(prepared)