fix(security): validate URLs before network requests in arxiv tool and devtools

Apply validate_url() before fetching the arXiv API and PyPI polling
endpoints. Replace urllib.request.urlopen with requests.get to satisfy
bandit S310 without noqa suppressions. Update arxiv tool tests to mock
requests.get and validate_url.

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

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

@@ -10,6 +10,9 @@ import xml.etree.ElementTree as ET
 
 from crewai.tools import BaseTool, EnvVar
 from pydantic import BaseModel, ConfigDict, Field
+import requests
+
+from crewai_tools.security.safe_path import validate_url
 
 
 logger = logging.getLogger(__file__)
@@ -78,17 +81,17 @@ class ArxivPaperTool(BaseTool):
     def fetch_arxiv_data(
         self, search_query: str, max_results: int
     ) -> list[dict[str, Any]]:
-        api_url = f"{self.BASE_API_URL}?search_query={urllib.parse.quote(search_query)}&start=0&max_results={max_results}"
+        api_url = validate_url(
+            f"{self.BASE_API_URL}?search_query={urllib.parse.quote(search_query)}"
+            f"&start=0&max_results={max_results}"
+        )
         logger.info(f"Fetching data from Arxiv API: {api_url}")
 
         try:
-            with urllib.request.urlopen(  # noqa: S310
-                api_url, timeout=self.REQUEST_TIMEOUT
-            ) as response:
-                if response.status != 200:
-                    raise Exception(f"HTTP {response.status}: {response.reason}")
-                data = response.read().decode("utf-8")
-        except urllib.error.URLError as e:
+            response = requests.get(api_url, timeout=self.REQUEST_TIMEOUT)
+            response.raise_for_status()
+            data = response.text
+        except requests.RequestException as e:
             logger.error(f"Error fetching data from Arxiv: {e}")
             raise
 

lib/crewai-tools/tests/tools/arxiv_paper_tool_test.py

@@ -1,10 +1,10 @@
 from pathlib import Path
 from unittest.mock import MagicMock, patch
-import urllib.error
 import xml.etree.ElementTree as ET
 
 from crewai_tools import ArxivPaperTool
 import pytest
+import requests
 
 
 @pytest.fixture
@@ -12,6 +12,15 @@ def tool():
     return ArxivPaperTool(download_pdfs=False)
 
 
+@pytest.fixture(autouse=True)
+def mock_validate_url():
+    with patch(
+        "crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.validate_url",
+        side_effect=lambda url: url,
+    ):
+        yield
+
+
 def mock_arxiv_response():
     return """<?xml version="1.0" encoding="UTF-8"?>
         <feed xmlns="http://www.w3.org/2005/Atom">
@@ -26,21 +35,23 @@ def mock_arxiv_response():
         </feed>"""
 
 
-@patch("urllib.request.urlopen")
-def test_fetch_arxiv_data(mock_urlopen, tool):
+@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
+def test_fetch_arxiv_data(mock_get, tool):
     mock_response = MagicMock()
-    mock_response.status = 200
-    mock_response.read.return_value = mock_arxiv_response().encode("utf-8")
-    mock_urlopen.return_value.__enter__.return_value = mock_response
+    mock_response.text = mock_arxiv_response()
+    mock_get.return_value = mock_response
 
     results = tool.fetch_arxiv_data("transformer", 1)
     assert isinstance(results, list)
     assert results[0]["title"] == "Sample Paper"
 
 
-@patch("urllib.request.urlopen", side_effect=urllib.error.URLError("Timeout"))
-def test_fetch_arxiv_data_network_error(mock_urlopen, tool):
-    with pytest.raises(urllib.error.URLError):
+@patch(
+    "crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get",
+    side_effect=requests.RequestException("Timeout"),
+)
+def test_fetch_arxiv_data_network_error(mock_get, tool):
+    with pytest.raises(requests.RequestException):
         tool.fetch_arxiv_data("transformer", 1)
 
 
@@ -60,13 +71,12 @@ def test_download_pdf_oserror(mock_urlretrieve):
         )
 
 
-@patch("urllib.request.urlopen")
+@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
 @patch("urllib.request.urlretrieve")
-def test_run_with_download(mock_urlretrieve, mock_urlopen):
+def test_run_with_download(mock_urlretrieve, mock_get):
     mock_response = MagicMock()
-    mock_response.status = 200
-    mock_response.read.return_value = mock_arxiv_response().encode("utf-8")
-    mock_urlopen.return_value.__enter__.return_value = mock_response
+    mock_response.text = mock_arxiv_response()
+    mock_get.return_value = mock_response
 
     tool = ArxivPaperTool(download_pdfs=True)
     output = tool._run("transformer", 1)
@@ -74,12 +84,11 @@ def test_run_with_download(mock_urlretrieve, mock_urlopen):
     mock_urlretrieve.assert_called_once()
 
 
-@patch("urllib.request.urlopen")
-def test_run_no_download(mock_urlopen):
+@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
+def test_run_no_download(mock_get):
     mock_response = MagicMock()
-    mock_response.status = 200
-    mock_response.read.return_value = mock_arxiv_response().encode("utf-8")
-    mock_urlopen.return_value.__enter__.return_value = mock_response
+    mock_response.text = mock_arxiv_response()
+    mock_get.return_value = mock_response
 
     tool = ArxivPaperTool(download_pdfs=False)
     result = tool._run("transformer", 1)
@@ -93,20 +102,19 @@ def test_validate_save_path_creates_directory(mock_mkdir):
     assert isinstance(path, Path)
 
 
-@patch("urllib.request.urlopen")
-def test_run_handles_exception(mock_urlopen):
-    mock_urlopen.side_effect = Exception("API failure")
+@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
+def test_run_handles_exception(mock_get):
+    mock_get.side_effect = Exception("API failure")
     tool = ArxivPaperTool()
     result = tool._run("transformer", 1)
     assert "Failed to fetch or download Arxiv papers" in result
 
 
-@patch("urllib.request.urlopen")
-def test_invalid_xml_response(mock_urlopen, tool):
+@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
+def test_invalid_xml_response(mock_get, tool):
     mock_response = MagicMock()
-    mock_response.read.return_value = b"<invalid><xml>"
-    mock_response.status = 200
-    mock_urlopen.return_value.__enter__.return_value = mock_response
+    mock_response.text = "<invalid><xml>"
+    mock_get.return_value = mock_response
 
     with pytest.raises(ET.ParseError):
         tool.fetch_arxiv_data("quantum", 1)

lib/devtools/pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "python-dotenv>=1.2.2,<2",
     "pygithub~=1.59.1",
     "rich>=13.9.4",
+    "crewai-tools",
 ]
 
 [project.scripts]

lib/devtools/src/crewai_devtools/cli.py

@@ -9,12 +9,13 @@ import sys
 import tempfile
 import time
 from typing import Final, Literal
-from urllib.request import urlopen
 
 import click
+from crewai_tools.security.safe_path import validate_url
 from dotenv import load_dotenv
 from github import Github
 from openai import OpenAI
+import requests
 from rich.console import Console
 from rich.markdown import Markdown
 from rich.panel import Panel
@@ -1548,14 +1549,14 @@ def _wait_for_pypi(package: str, version: str) -> None:
         package: PyPI package name.
         version: Version string to wait for.
     """
-    url = f"https://pypi.org/pypi/{package}/{version}/json"
+    url = validate_url(f"https://pypi.org/pypi/{package}/{version}/json")
     deadline = time.monotonic() + _PYPI_POLL_TIMEOUT
 
     console.print(f"[cyan]Waiting for {package}=={version} to appear on PyPI...[/cyan]")
     while time.monotonic() < deadline:
         try:
-            with urlopen(url) as resp:  # noqa: S310
-                if resp.status == 200:
+            response = requests.get(url, timeout=30)
+            if response.status_code == 200:
                     console.print(
                         f"[green]✓[/green] {package}=={version} is available on PyPI"
                     )