fix(memory): guard local uploads against credential reads

This commit is contained in:
dsad 2026-07-03 19:41:40 +03:00 committed by Teknium
parent d577408f3f
commit e02cef0d0d
4 changed files with 117 additions and 1 deletions

View file

@ -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:

View file

@ -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"