wehub-resource-sync
252 lines
9.0 KiB
Python
252 lines
9.0 KiB
Python
"""Tests for the auto-upload directory allowlist (parity with TS sdk)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from composio.core.models.base import allow_tracking
|
|
from composio.exceptions import (
|
|
FileUploadPathNotAllowedError,
|
|
SDKFileNotFoundError,
|
|
)
|
|
from composio.utils.upload_dir_allowlist import (
|
|
assert_path_inside_upload_dirs,
|
|
get_default_upload_dir,
|
|
resolve_effective_upload_allowlist,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def disable_telemetry():
|
|
token = allow_tracking.set(False)
|
|
yield
|
|
allow_tracking.reset(token)
|
|
|
|
|
|
class TestResolveEffectiveUploadAllowlist:
|
|
def test_none_returns_default(self):
|
|
result = resolve_effective_upload_allowlist(None)
|
|
default = get_default_upload_dir()
|
|
assert default is not None, "Expected a default upload dir in this env."
|
|
assert result == [default]
|
|
|
|
def test_empty_list_replaces_default(self):
|
|
assert resolve_effective_upload_allowlist([]) == []
|
|
|
|
def test_false_means_no_local_paths(self):
|
|
# Explicit "reject all local paths during auto-upload". Equivalent to []
|
|
# but reads better at call sites.
|
|
assert resolve_effective_upload_allowlist(False) == []
|
|
|
|
def test_explicit_list_replaces_default(self, tmp_path: Path):
|
|
result = resolve_effective_upload_allowlist([str(tmp_path)])
|
|
assert result == [tmp_path.resolve()]
|
|
|
|
def test_tilde_is_expanded(self):
|
|
result = resolve_effective_upload_allowlist(["~"])
|
|
assert result == [Path.home().resolve()]
|
|
|
|
def test_skips_blank_and_non_string_entries(self, tmp_path: Path):
|
|
result = resolve_effective_upload_allowlist([str(tmp_path), "", " "])
|
|
assert result == [tmp_path.resolve()]
|
|
|
|
def test_deduplicates_entries(self, tmp_path: Path):
|
|
result = resolve_effective_upload_allowlist([str(tmp_path), str(tmp_path)])
|
|
assert result == [tmp_path.resolve()]
|
|
|
|
|
|
class TestAssertPathInsideUploadDirs:
|
|
def test_accepts_direct_child(self, tmp_path: Path):
|
|
allowed = tmp_path
|
|
f = allowed / "a.txt"
|
|
f.write_text("hi")
|
|
assert_path_inside_upload_dirs(str(f), [allowed])
|
|
|
|
def test_accepts_nested_child(self, tmp_path: Path):
|
|
allowed = tmp_path
|
|
nested = allowed / "sub" / "deeper"
|
|
nested.mkdir(parents=True)
|
|
f = nested / "a.txt"
|
|
f.write_text("hi")
|
|
assert_path_inside_upload_dirs(str(f), [allowed])
|
|
|
|
def test_rejects_path_outside(self, tmp_path: Path):
|
|
allowed = tmp_path / "inside"
|
|
outside = tmp_path / "outside"
|
|
allowed.mkdir()
|
|
outside.mkdir()
|
|
f = outside / "a.txt"
|
|
f.write_text("hi")
|
|
with pytest.raises(FileUploadPathNotAllowedError):
|
|
assert_path_inside_upload_dirs(str(f), [allowed])
|
|
|
|
def test_enforces_component_boundary(self, tmp_path: Path):
|
|
# /tmp/foo must NOT accept /tmp/foo-bar/x.
|
|
allowed = tmp_path / "foo"
|
|
sibling = tmp_path / "foo-bar"
|
|
allowed.mkdir()
|
|
sibling.mkdir()
|
|
f = sibling / "x.txt"
|
|
f.write_text("hi")
|
|
with pytest.raises(FileUploadPathNotAllowedError):
|
|
assert_path_inside_upload_dirs(str(f), [allowed])
|
|
|
|
def test_raises_not_found_when_missing(self, tmp_path: Path):
|
|
allowed = tmp_path
|
|
missing = tmp_path / "does_not_exist.txt"
|
|
with pytest.raises(SDKFileNotFoundError) as excinfo:
|
|
assert_path_inside_upload_dirs(str(missing), [allowed])
|
|
assert "does not exist on disk" in str(excinfo.value)
|
|
assert "file_upload_dirs" in str(excinfo.value)
|
|
|
|
def test_empty_allowlist_fails_closed(self, tmp_path: Path):
|
|
f = tmp_path / "a.txt"
|
|
f.write_text("hi")
|
|
with pytest.raises(FileUploadPathNotAllowedError) as excinfo:
|
|
assert_path_inside_upload_dirs(str(f), [])
|
|
msg = str(excinfo.value)
|
|
assert "no upload directories are configured" in msg
|
|
assert "dangerously_allow_auto_upload_download_files" in msg
|
|
|
|
def test_error_message_lists_allowlist(self, tmp_path: Path):
|
|
allowed = tmp_path / "inside"
|
|
outside = tmp_path / "outside"
|
|
allowed.mkdir()
|
|
outside.mkdir()
|
|
f = outside / "a.txt"
|
|
f.write_text("hi")
|
|
with pytest.raises(FileUploadPathNotAllowedError) as excinfo:
|
|
assert_path_inside_upload_dirs(str(f), [allowed])
|
|
msg = str(excinfo.value)
|
|
assert str(allowed) in msg
|
|
assert "file_upload_dirs" in msg
|
|
assert "dangerously_allow_auto_upload_download_files" in msg
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only symlink semantics")
|
|
def test_rejects_symlink_pointing_outside(self, tmp_path: Path):
|
|
allowed = tmp_path / "inside"
|
|
outside = tmp_path / "outside"
|
|
allowed.mkdir()
|
|
outside.mkdir()
|
|
secret = outside / "secret.txt"
|
|
secret.write_text("pw")
|
|
link = allowed / "link.txt"
|
|
os.symlink(secret, link)
|
|
|
|
with pytest.raises(FileUploadPathNotAllowedError):
|
|
assert_path_inside_upload_dirs(str(link), [allowed])
|
|
|
|
|
|
class TestToolsWiresAllowlistCorrectly:
|
|
"""Verifies ``Tools`` forwards the right allowlist to ``FileHelper`` given
|
|
the full flag/value matrix."""
|
|
|
|
@staticmethod
|
|
def _make_tools(
|
|
dangerously_allow_auto_upload_download_files: bool,
|
|
file_upload_dirs,
|
|
):
|
|
from composio.core.models.tools import Tools
|
|
from unittest.mock import Mock
|
|
|
|
return Tools(
|
|
client=Mock(),
|
|
provider=Mock(name="provider"),
|
|
dangerously_allow_auto_upload_download_files=dangerously_allow_auto_upload_download_files,
|
|
file_upload_dirs=file_upload_dirs,
|
|
)
|
|
|
|
def test_flag_off_means_no_allowlist_is_ever_built(self):
|
|
# Flag off ⇒ auto-upload code path is never run. We pass ``None`` to
|
|
# FileHelper so manual/no-op paths don't accidentally enforce anything.
|
|
t = self._make_tools(False, None)
|
|
assert t._file_helper._file_upload_allowlist is None # noqa: SLF001
|
|
|
|
def test_flag_off_ignores_file_upload_dirs_entirely(self):
|
|
t = self._make_tools(False, ["/tmp/whatever"])
|
|
assert t._file_helper._file_upload_allowlist is None # noqa: SLF001
|
|
|
|
t = self._make_tools(False, False)
|
|
assert t._file_helper._file_upload_allowlist is None # noqa: SLF001
|
|
|
|
t = self._make_tools(False, [])
|
|
assert t._file_helper._file_upload_allowlist is None # noqa: SLF001
|
|
|
|
def test_flag_on_none_uses_default(self):
|
|
t = self._make_tools(True, None)
|
|
assert t._file_helper._file_upload_allowlist == [ # noqa: SLF001
|
|
get_default_upload_dir()
|
|
]
|
|
|
|
def test_flag_on_false_means_empty_allowlist(self):
|
|
t = self._make_tools(True, False)
|
|
assert t._file_helper._file_upload_allowlist == [] # noqa: SLF001
|
|
|
|
def test_flag_on_empty_list_means_empty_allowlist(self):
|
|
t = self._make_tools(True, [])
|
|
assert t._file_helper._file_upload_allowlist == [] # noqa: SLF001
|
|
|
|
def test_flag_on_explicit_dirs_replace_default(self, tmp_path: Path):
|
|
t = self._make_tools(True, [str(tmp_path)])
|
|
assert t._file_helper._file_upload_allowlist == [ # noqa: SLF001
|
|
tmp_path.resolve()
|
|
]
|
|
|
|
|
|
class TestFileUploadableAllowlistWiring:
|
|
"""Sanity: FileUploadable.from_path must honor the allowlist param."""
|
|
|
|
def test_from_path_with_allowlist_blocks_outside(self, tmp_path: Path):
|
|
from composio.core.models._files import FileUploadable
|
|
from unittest.mock import Mock
|
|
|
|
inside = tmp_path / "inside"
|
|
outside = tmp_path / "outside"
|
|
inside.mkdir()
|
|
outside.mkdir()
|
|
f = outside / "a.txt"
|
|
f.write_text("hi")
|
|
|
|
with pytest.raises(FileUploadPathNotAllowedError):
|
|
FileUploadable.from_path(
|
|
client=Mock(),
|
|
file=str(f),
|
|
tool="T",
|
|
toolkit="tk",
|
|
file_upload_allowlist=[inside],
|
|
)
|
|
|
|
def test_from_path_without_allowlist_skips_check(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""Passing ``file_upload_allowlist=None`` must NOT trigger the check.
|
|
|
|
This simulates manual upload paths and the no-auto-upload default.
|
|
"""
|
|
from composio.core.models import _files as files_mod
|
|
from composio.core.models._files import FileUploadable
|
|
from unittest.mock import Mock
|
|
|
|
f = tmp_path / "a.txt"
|
|
f.write_text("hi")
|
|
|
|
# Short-circuit the network-bound parts.
|
|
mock_client = Mock()
|
|
mock_client.post.return_value = Mock(
|
|
new_presigned_url="https://example/upload", key="k"
|
|
)
|
|
monkeypatch.setattr(files_mod, "upload", lambda url, file: True)
|
|
|
|
result = FileUploadable.from_path(
|
|
client=mock_client,
|
|
file=str(f),
|
|
tool="T",
|
|
toolkit="tk",
|
|
file_upload_allowlist=None,
|
|
)
|
|
assert result.s3key == "k"
|