diff --git a/lib/crewai-tools/src/crewai_tools/tools/selenium_scraping_tool/selenium_scraping_tool.py b/lib/crewai-tools/src/crewai_tools/tools/selenium_scraping_tool/selenium_scraping_tool.py index 41ca86671e..41891f23bf 100644 --- a/lib/crewai-tools/src/crewai_tools/tools/selenium_scraping_tool/selenium_scraping_tool.py +++ b/lib/crewai-tools/src/crewai_tools/tools/selenium_scraping_tool/selenium_scraping_tool.py @@ -6,6 +6,8 @@ from crewai.tools import BaseTool from pydantic import BaseModel, Field, field_validator +from crewai_tools.security.safe_path import validate_url + class FixedSeleniumScrapingToolSchema(BaseModel): """Input for SeleniumScrapingTool.""" @@ -45,7 +47,9 @@ def validate_website_url(cls, v: str) -> str: if re.search(r"\s", v): raise ValueError("URL cannot contain whitespace") - return v + # Align with Firecrawl/Scrapfly: block private/reserved SSRF targets + # before the local Chrome WebDriver navigates to the URL. + return validate_url(v) class SeleniumScrapingTool(BaseTool): @@ -121,9 +125,9 @@ def __init__( self.css_element = css_element if website_url is not None: - self.website_url = website_url + self.website_url = validate_url(website_url) self.description = ( - f"A tool that can be used to read {website_url}'s content." + f"A tool that can be used to read {self.website_url}'s content." ) self.args_schema = FixedSeleniumScrapingToolSchema @@ -195,6 +199,9 @@ def _make_request( if not re.match(r"^https?://", url): raise ValueError("URL must start with http:// or https://") + # Defense in depth for the fixed-URL schema path (no pydantic re-check). + url = validate_url(url) + if self.driver is None: raise RuntimeError("Driver not initialized. Call _run first.") sleep_time = wait_time or 0 diff --git a/lib/crewai-tools/tests/tools/selenium_scraping_tool_test.py b/lib/crewai-tools/tests/tools/selenium_scraping_tool_test.py index c60629453a..9f284c9a4e 100644 --- a/lib/crewai-tools/tests/tools/selenium_scraping_tool_test.py +++ b/lib/crewai-tools/tests/tools/selenium_scraping_tool_test.py @@ -129,3 +129,37 @@ def test_initialization_with_driver(_mocked_chrome_driver): mock_driver = MagicMock() tool = initialize_tool_with(mock_driver) assert tool.driver == mock_driver + + +@patch("selenium.webdriver.Chrome") +def test_rejects_loopback_ssrf_targets(_mocked_chrome_driver): + mock_driver = mock_driver_with_html("nope") + tool = initialize_tool_with(mock_driver) + + result = tool._run(website_url="http://127.0.0.1/admin") + + assert "private/reserved" in result.lower() or "Error scraping website" in result + mock_driver.get.assert_not_called() + + +@patch("selenium.webdriver.Chrome") +def test_rejects_cloud_metadata_ssrf_targets(_mocked_chrome_driver): + mock_driver = mock_driver_with_html("nope") + tool = initialize_tool_with(mock_driver) + + result = tool._run(website_url="http://169.254.169.254/latest/meta-data/") + + assert "private/reserved" in result.lower() or "Error scraping website" in result + mock_driver.get.assert_not_called() + + +@patch("selenium.webdriver.Chrome") +def test_fixed_url_constructor_rejects_private_targets(mocked_chrome): + mocked_chrome.return_value = MagicMock() + try: + SeleniumScrapingTool(website_url="http://localhost/internal") + raised = False + except ValueError as exc: + raised = True + assert "private/reserved" in str(exc).lower() or "localhost" in str(exc).lower() + assert raised