mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-04-14 06:53:25 +00:00
57e23b4dca
- Replace insecure tempfile.mktemp() with inline symlink target in test - Remove unused 'target' variable and unused tempfile import - Narrow broad except Exception: pass to only catch urlparse errors; validate_url ValueError now propagates instead of being silently swallowed - Fix ruff B904 (raise-without-from-inside-except) in safe_path.py - Fix ruff B007 (unused loop variable 'family') in safe_path.py - Use validate_directory_path in DirectorySearchTool.add() so the public utility is exercised in production code Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
171 lines
6.7 KiB
Python
171 lines
6.7 KiB
Python
"""Tests for path and URL validation utilities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from crewai_tools.utilities.safe_path import (
|
|
validate_directory_path,
|
|
validate_file_path,
|
|
validate_url,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# File path validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestValidateFilePath:
|
|
"""Tests for validate_file_path."""
|
|
|
|
def test_valid_relative_path(self, tmp_path):
|
|
"""Normal relative path within the base directory."""
|
|
(tmp_path / "data.json").touch()
|
|
result = validate_file_path("data.json", str(tmp_path))
|
|
assert result == str(tmp_path / "data.json")
|
|
|
|
def test_valid_nested_path(self, tmp_path):
|
|
"""Nested path within base directory."""
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "file.txt").touch()
|
|
result = validate_file_path("sub/file.txt", str(tmp_path))
|
|
assert result == str(tmp_path / "sub" / "file.txt")
|
|
|
|
def test_rejects_dotdot_traversal(self, tmp_path):
|
|
"""Reject ../ traversal that escapes base_dir."""
|
|
with pytest.raises(ValueError, match="outside the allowed directory"):
|
|
validate_file_path("../../etc/passwd", str(tmp_path))
|
|
|
|
def test_rejects_absolute_path_outside_base(self, tmp_path):
|
|
"""Reject absolute path outside base_dir."""
|
|
with pytest.raises(ValueError, match="outside the allowed directory"):
|
|
validate_file_path("/etc/passwd", str(tmp_path))
|
|
|
|
def test_allows_absolute_path_inside_base(self, tmp_path):
|
|
"""Allow absolute path that's inside base_dir."""
|
|
(tmp_path / "ok.txt").touch()
|
|
result = validate_file_path(str(tmp_path / "ok.txt"), str(tmp_path))
|
|
assert result == str(tmp_path / "ok.txt")
|
|
|
|
def test_rejects_symlink_escape(self, tmp_path):
|
|
"""Reject symlinks that point outside base_dir."""
|
|
link = tmp_path / "sneaky_link"
|
|
# Create a symlink pointing to /etc/passwd
|
|
os.symlink("/etc/passwd", str(link))
|
|
with pytest.raises(ValueError, match="outside the allowed directory"):
|
|
validate_file_path("sneaky_link", str(tmp_path))
|
|
|
|
def test_defaults_to_cwd(self):
|
|
"""When no base_dir is given, use cwd."""
|
|
cwd = os.getcwd()
|
|
# A file in cwd should be valid
|
|
result = validate_file_path(".", None)
|
|
assert result == os.path.realpath(cwd)
|
|
|
|
def test_escape_hatch(self, tmp_path, monkeypatch):
|
|
"""CREWAI_TOOLS_ALLOW_UNSAFE_PATHS=true bypasses validation."""
|
|
monkeypatch.setenv("CREWAI_TOOLS_ALLOW_UNSAFE_PATHS", "true")
|
|
# This would normally be rejected
|
|
result = validate_file_path("/etc/passwd", str(tmp_path))
|
|
assert result == os.path.realpath("/etc/passwd")
|
|
|
|
|
|
class TestValidateDirectoryPath:
|
|
"""Tests for validate_directory_path."""
|
|
|
|
def test_valid_directory(self, tmp_path):
|
|
(tmp_path / "subdir").mkdir()
|
|
result = validate_directory_path("subdir", str(tmp_path))
|
|
assert result == str(tmp_path / "subdir")
|
|
|
|
def test_rejects_file_as_directory(self, tmp_path):
|
|
(tmp_path / "file.txt").touch()
|
|
with pytest.raises(ValueError, match="not a directory"):
|
|
validate_directory_path("file.txt", str(tmp_path))
|
|
|
|
def test_rejects_traversal(self, tmp_path):
|
|
with pytest.raises(ValueError, match="outside the allowed directory"):
|
|
validate_directory_path("../../", str(tmp_path))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# URL validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestValidateUrl:
|
|
"""Tests for validate_url."""
|
|
|
|
def test_valid_https_url(self):
|
|
"""Normal HTTPS URL should pass."""
|
|
result = validate_url("https://example.com/data.json")
|
|
assert result == "https://example.com/data.json"
|
|
|
|
def test_valid_http_url(self):
|
|
"""Normal HTTP URL should pass."""
|
|
result = validate_url("http://example.com/api")
|
|
assert result == "http://example.com/api"
|
|
|
|
def test_blocks_file_scheme(self):
|
|
"""file:// URLs must be blocked."""
|
|
with pytest.raises(ValueError, match="file:// URLs are not allowed"):
|
|
validate_url("file:///etc/passwd")
|
|
|
|
def test_blocks_file_scheme_with_host(self):
|
|
with pytest.raises(ValueError, match="file:// URLs are not allowed"):
|
|
validate_url("file://localhost/etc/shadow")
|
|
|
|
def test_blocks_localhost(self):
|
|
"""localhost must be blocked (resolves to 127.0.0.1)."""
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://localhost/admin")
|
|
|
|
def test_blocks_127_0_0_1(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://127.0.0.1/admin")
|
|
|
|
def test_blocks_cloud_metadata(self):
|
|
"""AWS/GCP/Azure metadata endpoint must be blocked."""
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://169.254.169.254/latest/meta-data/")
|
|
|
|
def test_blocks_private_10_range(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://10.0.0.1/internal")
|
|
|
|
def test_blocks_private_172_range(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://172.16.0.1/internal")
|
|
|
|
def test_blocks_private_192_range(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://192.168.1.1/router")
|
|
|
|
def test_blocks_zero_address(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://0.0.0.0/")
|
|
|
|
def test_blocks_ipv6_localhost(self):
|
|
with pytest.raises(ValueError, match="private/reserved IP"):
|
|
validate_url("http://[::1]/admin")
|
|
|
|
def test_blocks_ftp_scheme(self):
|
|
with pytest.raises(ValueError, match="not allowed"):
|
|
validate_url("ftp://example.com/file")
|
|
|
|
def test_blocks_empty_hostname(self):
|
|
with pytest.raises(ValueError, match="no hostname"):
|
|
validate_url("http:///path")
|
|
|
|
def test_blocks_unresolvable_host(self):
|
|
with pytest.raises(ValueError, match="Could not resolve"):
|
|
validate_url("http://this-host-definitely-does-not-exist-abc123.com/")
|
|
|
|
def test_escape_hatch(self, monkeypatch):
|
|
"""CREWAI_TOOLS_ALLOW_UNSAFE_PATHS=true bypasses URL validation."""
|
|
monkeypatch.setenv("CREWAI_TOOLS_ALLOW_UNSAFE_PATHS", "true")
|
|
# file:// would normally be blocked
|
|
result = validate_url("file:///etc/passwd")
|
|
assert result == "file:///etc/passwd"
|