mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-22 07:15:10 +00:00
test(security): cover validate_url usage in arxiv tool and PyPI polling
Add behavior tests that assert URL validation runs before network requests and that unsafe URLs never reach requests.get. Also fix indentation in _wait_for_pypi after the requests migration. Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>
This commit is contained in:
@@ -136,3 +136,51 @@ def test_run_with_max_results(mock_fetch, tool):
|
||||
|
||||
result = tool._run(search_query="test", max_results=100)
|
||||
assert result.count("Title:") == 100
|
||||
|
||||
|
||||
@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
|
||||
@patch(
|
||||
"crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.validate_url",
|
||||
return_value="https://validated.example/api/query?search_query=transformer&start=0&max_results=1",
|
||||
)
|
||||
def test_fetch_arxiv_data_validates_url_before_request(mock_validate_url, mock_get, tool):
|
||||
mock_response = MagicMock()
|
||||
mock_response.text = mock_arxiv_response()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
tool.fetch_arxiv_data("transformer", 1)
|
||||
|
||||
mock_validate_url.assert_called_once()
|
||||
validated_url = mock_validate_url.call_args.args[0]
|
||||
assert validated_url.startswith(ArxivPaperTool.BASE_API_URL)
|
||||
assert "search_query=transformer" in validated_url
|
||||
assert "max_results=1" in validated_url
|
||||
mock_get.assert_called_once_with(
|
||||
"https://validated.example/api/query?search_query=transformer&start=0&max_results=1",
|
||||
timeout=ArxivPaperTool.REQUEST_TIMEOUT,
|
||||
)
|
||||
|
||||
|
||||
@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
|
||||
@patch(
|
||||
"crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.validate_url",
|
||||
side_effect=ValueError("URL resolves to private/reserved IP"),
|
||||
)
|
||||
def test_fetch_arxiv_data_rejects_unsafe_url(mock_validate_url, mock_get, tool):
|
||||
with pytest.raises(ValueError, match="private/reserved IP"):
|
||||
tool.fetch_arxiv_data("transformer", 1)
|
||||
|
||||
mock_validate_url.assert_called_once()
|
||||
mock_get.assert_not_called()
|
||||
|
||||
|
||||
@patch("crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.requests.get")
|
||||
@patch(
|
||||
"crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool.validate_url",
|
||||
side_effect=ValueError("URL resolves to private/reserved IP"),
|
||||
)
|
||||
def test_run_returns_error_when_url_validation_fails(mock_validate_url, mock_get, tool):
|
||||
result = tool._run("transformer", 1)
|
||||
|
||||
assert "Failed to fetch or download Arxiv papers" in result
|
||||
mock_get.assert_not_called()
|
||||
|
||||
@@ -1557,10 +1557,10 @@ def _wait_for_pypi(package: str, version: str) -> None:
|
||||
try:
|
||||
response = requests.get(url, timeout=30)
|
||||
if response.status_code == 200:
|
||||
console.print(
|
||||
f"[green]✓[/green] {package}=={version} is available on PyPI"
|
||||
)
|
||||
return
|
||||
console.print(
|
||||
f"[green]✓[/green] {package}=={version} is available on PyPI"
|
||||
)
|
||||
return
|
||||
except Exception: # noqa: S110
|
||||
pass
|
||||
time.sleep(_PYPI_POLL_INTERVAL)
|
||||
|
||||
37
lib/devtools/tests/test_wait_for_pypi.py
Normal file
37
lib/devtools/tests/test_wait_for_pypi.py
Normal file
@@ -0,0 +1,37 @@
|
||||
"""Tests for PyPI polling URL validation in the release CLI."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from crewai_devtools.cli import _wait_for_pypi
|
||||
import pytest
|
||||
|
||||
|
||||
@patch("crewai_devtools.cli.requests.get")
|
||||
@patch(
|
||||
"crewai_devtools.cli.validate_url",
|
||||
return_value="https://pypi.org/pypi/crewai/1.0.0/json",
|
||||
)
|
||||
def test_wait_for_pypi_validates_url_before_request(mock_validate_url, mock_get):
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
_wait_for_pypi("crewai", "1.0.0")
|
||||
|
||||
mock_validate_url.assert_called_once_with("https://pypi.org/pypi/crewai/1.0.0/json")
|
||||
mock_get.assert_called_once_with(
|
||||
"https://pypi.org/pypi/crewai/1.0.0/json", timeout=30
|
||||
)
|
||||
|
||||
|
||||
@patch("crewai_devtools.cli.requests.get")
|
||||
@patch(
|
||||
"crewai_devtools.cli.validate_url",
|
||||
side_effect=ValueError("URL resolves to private/reserved IP"),
|
||||
)
|
||||
def test_wait_for_pypi_rejects_unsafe_url(mock_validate_url, mock_get):
|
||||
with pytest.raises(ValueError, match="private/reserved IP"):
|
||||
_wait_for_pypi("crewai", "1.0.0")
|
||||
|
||||
mock_validate_url.assert_called_once()
|
||||
mock_get.assert_not_called()
|
||||
Reference in New Issue
Block a user