diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index c6ea6bc1d7c..5aef3b90850 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -576,6 +576,8 @@ _TOOL_STATUS_COMPLETED_ALIASES = {"completed", "complete", "success", "succeeded def _zip_directory(dir_path: Path) -> Path: """Create a temporary zip file containing a directory tree.""" + from agent.file_safety import raise_if_read_blocked + root = dir_path.resolve() zip_path = Path(tempfile.gettempdir()) / f"openviking_upload_{uuid.uuid4().hex}.zip" with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf: @@ -584,7 +586,12 @@ def _zip_directory(dir_path: Path) -> Path: continue if file_path.is_file(): try: - file_path.resolve().relative_to(root) + resolved = file_path.resolve() + resolved.relative_to(root) + except ValueError: + continue + try: + raise_if_read_blocked(str(resolved)) except ValueError: continue arcname = str(file_path.relative_to(dir_path)).replace("\\", "/") @@ -3645,6 +3652,8 @@ class OpenVikingMemoryProvider(MemoryProvider): return json.dumps(payload, ensure_ascii=False) def _tool_add_resource(self, args: dict) -> str: + from agent.file_safety import raise_if_read_blocked + url = args.get("url", "") if not url: return tool_error("url is required") @@ -3678,6 +3687,10 @@ class OpenVikingMemoryProvider(MemoryProvider): cleanup_path = _zip_directory(source_path) upload_path = cleanup_path elif source_path.is_file(): + try: + raise_if_read_blocked(str(source_path)) + except ValueError as exc: + return tool_error(str(exc)) payload["source_name"] = source_path.name upload_path = source_path else: diff --git a/plugins/memory/retaindb/__init__.py b/plugins/memory/retaindb/__init__.py index 62121410d41..a777ccbd55e 100644 --- a/plugins/memory/retaindb/__init__.py +++ b/plugins/memory/retaindb/__init__.py @@ -34,6 +34,7 @@ from typing import Any, Dict, List from urllib.parse import quote from agent.memory_provider import MemoryProvider +from agent.file_safety import raise_if_read_blocked from tools.registry import tool_error logger = logging.getLogger(__name__) @@ -702,6 +703,10 @@ class RetainDBMemoryProvider(MemoryProvider): path_obj = Path(local_path) if not path_obj.exists(): return {"error": f"File not found: {local_path}"} + try: + raise_if_read_blocked(str(path_obj)) + except ValueError as exc: + return {"error": str(exc)} data = path_obj.read_bytes() import mimetypes mime = mimetypes.guess_type(path_obj.name)[0] or "application/octet-stream" diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index f8991e3e766..98a99755bfc 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -1298,6 +1298,26 @@ def test_tool_add_resource_uploads_file_uri(tmp_path): assert result["root_uri"] == "viking://resources/sample" +def test_tool_add_resource_rejects_hermes_credential_file_upload(tmp_path, monkeypatch): + import agent.file_safety as fs + + hermes_home = tmp_path / "hermes_home" + hermes_home.mkdir() + auth_json = hermes_home / "auth.json" + auth_json.write_text('{"OPENROUTER_API_KEY":"sk-test-secret"}', encoding="utf-8") + monkeypatch.setattr(fs, "_hermes_home_path", lambda: hermes_home) + + provider = OpenVikingMemoryProvider() + provider._client = MagicMock() + + result = json.loads(provider._tool_add_resource({"url": str(auth_json)})) + + assert "error" in result + assert "credential store" in result["error"] + provider._client.upload_temp_file.assert_not_called() + provider._client.post.assert_not_called() + + def test_tool_add_resource_uploads_existing_local_directory_and_cleans_zip(tmp_path): docs = tmp_path / "docs" docs.mkdir() @@ -1372,6 +1392,44 @@ def test_tool_add_resource_directory_zip_skips_symlink_escape(tmp_path): assert b"do not upload" not in b"".join(archive_entries["payloads"].values()) +def test_tool_add_resource_directory_zip_skips_hermes_credential_files(tmp_path, monkeypatch): + import agent.file_safety as fs + + hermes_home = tmp_path / "hermes_home" + hermes_home.mkdir() + (hermes_home / "guide.md").write_text("# Guide\n", encoding="utf-8") + (hermes_home / "auth.json").write_text( + '{"OPENROUTER_API_KEY":"sk-test-secret"}', + encoding="utf-8", + ) + monkeypatch.setattr(fs, "_hermes_home_path", lambda: hermes_home) + + provider = OpenVikingMemoryProvider() + provider._client = MagicMock() + archive_entries = {} + + def inspect_upload(path): + with zipfile.ZipFile(path) as archive: + archive_entries["names"] = archive.namelist() + archive_entries["payloads"] = { + name: archive.read(name) + for name in archive.namelist() + } + return "upload_hermes_home.zip" + + provider._client.upload_temp_file.side_effect = inspect_upload + provider._client.post.return_value = { + "status": "ok", + "result": {"root_uri": "viking://resources/hermes_home"}, + } + + result = json.loads(provider._tool_add_resource({"url": str(hermes_home)})) + + assert result["status"] == "added" + assert archive_entries["names"] == ["guide.md"] + assert b"sk-test-secret" not in b"".join(archive_entries["payloads"].values()) + + def test_tool_add_resource_cleans_local_directory_zip_when_add_fails(tmp_path): docs = tmp_path / "docs" docs.mkdir() diff --git a/tests/plugins/memory/test_retaindb_provider.py b/tests/plugins/memory/test_retaindb_provider.py new file mode 100644 index 00000000000..bc468980448 --- /dev/null +++ b/tests/plugins/memory/test_retaindb_provider.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from unittest.mock import MagicMock + +import agent.file_safety as fs + +from plugins.memory.retaindb import RetainDBMemoryProvider + + +def test_upload_file_rejects_hermes_credential_store(tmp_path, monkeypatch): + hermes_home = tmp_path / "hermes_home" + hermes_home.mkdir() + auth_json = hermes_home / "auth.json" + auth_json.write_text('{"OPENAI_API_KEY":"sk-test-secret"}', encoding="utf-8") + monkeypatch.setattr(fs, "_hermes_home_path", lambda: hermes_home) + + provider = RetainDBMemoryProvider() + provider._client = MagicMock() + + result = provider._dispatch("retaindb_upload_file", {"local_path": str(auth_json)}) + + assert "error" in result + assert "credential store" in result["error"] + provider._client.upload_file.assert_not_called() + + +def test_upload_file_allows_regular_file(tmp_path): + note = tmp_path / "note.md" + note.write_text("# Note\n", encoding="utf-8") + provider = RetainDBMemoryProvider() + provider._client = MagicMock() + provider._client.upload_file.return_value = { + "file": {"id": "file-1", "name": "note.md"}, + } + + result = provider._dispatch("retaindb_upload_file", {"local_path": str(note)}) + + provider._client.upload_file.assert_called_once() + assert provider._client.upload_file.call_args.args[0] == note.read_bytes() + assert result["file"]["id"] == "file-1"