fix: 修复沙盒共享权限配置错误的问题 #594

Author: xerrors

backend/package/yuxi/agents/backends/composite.py

@@ -94,6 +94,23 @@ def _extract_thread_id(runtime) -> str:
     raise ValueError("thread_id is required in runtime configurable context")
 
 
+def _extract_user_id(runtime) -> str:
+    config = getattr(runtime, "config", None)
+    if isinstance(config, dict):
+        configurable = config.get("configurable", {})
+        if isinstance(configurable, dict):
+            user_id = configurable.get("user_id")
+            if isinstance(user_id, str) and user_id.strip():
+                return user_id.strip()
+
+    context = getattr(runtime, "context", None)
+    user_id = getattr(context, "user_id", None)
+    if isinstance(user_id, str) and user_id.strip():
+        return user_id.strip()
+
+    raise ValueError("user_id is required in runtime configurable context")
+
+
 def _get_visible_knowledge_bases_from_runtime(runtime) -> list[dict]:
     context = getattr(runtime, "context", None)
     selected = getattr(context, "_visible_knowledge_bases", None)
@@ -105,9 +122,10 @@ def _get_visible_knowledge_bases_from_runtime(runtime) -> list[dict]:
 def create_agent_composite_backend(runtime) -> CompositeBackend:
     visible_skills = _get_visible_skills_from_runtime(runtime)
     thread_id = _extract_thread_id(runtime)
+    user_id = _extract_user_id(runtime)
     visible_kbs = _get_visible_knowledge_bases_from_runtime(runtime)
     return CustomCompositeBackend(
-        default=ProvisionerSandboxBackend(thread_id=thread_id, visible_skills=visible_skills),
+        default=ProvisionerSandboxBackend(thread_id=thread_id, user_id=user_id, visible_skills=visible_skills),
         routes={
             "/skills/": SelectedSkillsReadonlyBackend(selected_slugs=visible_skills),
             "/home/gem/kbs/": KnowledgeBaseReadonlyBackend(visible_kbs=visible_kbs),

backend/package/yuxi/agents/backends/sandbox/backend.py

@@ -62,10 +62,13 @@ def _looks_like_binary(content: bytes) -> bool:
 
 
 class ProvisionerSandboxBackend(BaseSandbox):
-    def __init__(self, thread_id: str, *, visible_skills: list[str] | None = None):
+    def __init__(self, thread_id: str, *, user_id: str, visible_skills: list[str] | None = None):
         self._thread_id = str(thread_id or "").strip()
         if not self._thread_id:
             raise ValueError("thread_id is required for ProvisionerSandboxBackend")
+        self._user_id = str(user_id or "").strip()
+        if not self._user_id:
+            raise ValueError("user_id is required for ProvisionerSandboxBackend")
 
         self._visible_skills = list(visible_skills or [])
         self._provider = get_sandbox_provider()
@@ -91,7 +94,7 @@ def _build_client(self, sandbox_url: str):
 
     def _get_client(self) -> Any:
         sync_thread_visible_skills(self._thread_id, self._visible_skills)
-        connection = self._provider.get(self._thread_id, create_if_missing=True)
+        connection = self._provider.get(self._thread_id, user_id=self._user_id, create_if_missing=True)
         if connection is None:
             raise RuntimeError(f"sandbox is unavailable for thread {self._thread_id}")
 

backend/package/yuxi/agents/backends/sandbox/paths.py

@@ -6,7 +6,7 @@
 from yuxi import config as conf
 from yuxi.utils.paths import OUTPUTS_DIR_NAME, UPLOADS_DIR_NAME, VIRTUAL_PATH_PREFIX, WORKSPACE_DIR_NAME
 
-_SAFE_THREAD_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
 
 
 def get_virtual_path_prefix() -> str:
@@ -17,7 +17,7 @@ def _validate_thread_id(thread_id: str) -> str:
     value = str(thread_id or "").strip()
     if not value:
         raise ValueError("thread_id is required")
-    if not _SAFE_THREAD_ID_RE.match(value):
+    if not _SAFE_ID_RE.match(value):
         raise ValueError("thread_id contains invalid characters")
     return value
 
@@ -27,18 +27,28 @@ def _thread_root_dir(thread_id: str) -> Path:
     return Path(conf.save_dir) / "threads" / safe_thread_id / "user-data"
 
 
-def _global_user_data_dir() -> Path:
-    """Return the shared host-side directory used for thread workspace files."""
-    return Path(conf.save_dir) / "threads" / "shared"
+def _validate_user_id(user_id: str) -> str:
+    value = str(user_id or "").strip()
+    if not value:
+        raise ValueError("user_id is required")
+    if not _SAFE_ID_RE.match(value):
+        raise ValueError("user_id contains invalid characters")
+    return value
+
+
+def _global_user_data_dir(user_id: str) -> Path:
+    """Return the shared host-side directory used for one user's workspace files."""
+    safe_user_id = _validate_user_id(user_id)
+    return Path(conf.save_dir) / "threads" / "shared" / safe_user_id
 
 
 def sandbox_user_data_dir(thread_id: str) -> Path:
     return _thread_root_dir(thread_id)
 
 
-def sandbox_workspace_dir(thread_id: str) -> Path:
+def sandbox_workspace_dir(thread_id: str, user_id: str) -> Path:
     _validate_thread_id(thread_id)
-    return _global_user_data_dir() / WORKSPACE_DIR_NAME
+    return _global_user_data_dir(user_id) / WORKSPACE_DIR_NAME
 
 
 def sandbox_uploads_dir(thread_id: str) -> Path:
@@ -49,14 +59,14 @@ def sandbox_outputs_dir(thread_id: str) -> Path:
     return _thread_root_dir(thread_id) / OUTPUTS_DIR_NAME
 
 
-def ensure_thread_dirs(thread_id: str) -> None:
-    _global_user_data_dir().mkdir(parents=True, exist_ok=True)
-    sandbox_workspace_dir(thread_id).mkdir(parents=True, exist_ok=True)
+def ensure_thread_dirs(thread_id: str, user_id: str) -> None:
+    _global_user_data_dir(user_id).mkdir(parents=True, exist_ok=True)
+    sandbox_workspace_dir(thread_id, user_id).mkdir(parents=True, exist_ok=True)
     sandbox_uploads_dir(thread_id).mkdir(parents=True, exist_ok=True)
     sandbox_outputs_dir(thread_id).mkdir(parents=True, exist_ok=True)
 
 
-def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Path, Path]:
+def _resolve_user_data_base_dir(thread_id: str, user_id: str, relative_path: str) -> tuple[Path, Path]:
     """Map a virtual user-data path to the correct host-side base directory."""
     parts = Path(relative_path).parts
     if not parts:
@@ -65,8 +75,8 @@ def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Pat
 
     namespace = parts[0]
     if namespace == WORKSPACE_DIR_NAME:
-        # Workspace is shared across threads, so it lives outside the per-thread root.
-        base_dir = sandbox_workspace_dir(thread_id)
+        # Workspace is shared across one user's threads, so it lives outside the per-thread root.
+        base_dir = sandbox_workspace_dir(thread_id, user_id)
         target_path = base_dir.joinpath(*parts[1:]) if len(parts) > 1 else base_dir
         return base_dir.resolve(), target_path.resolve()
     if namespace == UPLOADS_DIR_NAME:
@@ -82,15 +92,15 @@ def _resolve_user_data_base_dir(thread_id: str, relative_path: str) -> tuple[Pat
     return base_dir.resolve(), (base_dir / relative_path).resolve()
 
 
-def resolve_virtual_path(thread_id: str, virtual_path: str) -> Path:
+def resolve_virtual_path(thread_id: str, virtual_path: str, *, user_id: str) -> Path:
     clean_virtual_path = "/" + str(virtual_path or "").strip().lstrip("/")
     virtual_prefix = get_virtual_path_prefix()
 
     if clean_virtual_path != virtual_prefix and not clean_virtual_path.startswith(f"{virtual_prefix}/"):
         raise ValueError(f"path must start with {virtual_prefix}")
 
     relative_path = clean_virtual_path[len(virtual_prefix) :].lstrip("/")
-    base_dir, target_path = _resolve_user_data_base_dir(thread_id, relative_path)
+    base_dir, target_path = _resolve_user_data_base_dir(thread_id, user_id, relative_path)
 
     try:
         target_path.relative_to(base_dir)
@@ -100,10 +110,10 @@ def resolve_virtual_path(thread_id: str, virtual_path: str) -> Path:
     return target_path
 
 
-def virtual_path_for_thread_file(thread_id: str, path: str | Path) -> str:
+def virtual_path_for_thread_file(thread_id: str, path: str | Path, *, user_id: str) -> str:
     target_path = Path(path).resolve()
     thread_root = sandbox_user_data_dir(thread_id).resolve()
-    global_workspace_root = sandbox_workspace_dir(thread_id).resolve()
+    global_workspace_root = sandbox_workspace_dir(thread_id, user_id).resolve()
 
     try:
         relative_path = target_path.relative_to(global_workspace_root)

backend/package/yuxi/agents/backends/sandbox/provider.py

@@ -19,6 +19,7 @@ def sandbox_id_for_thread(thread_id: str) -> str:
 @dataclass(slots=True)
 class SandboxConnection:
     thread_id: str
+    user_id: str
     sandbox_id: str
     sandbox_url: str
 
@@ -48,9 +49,10 @@ def _thread_lock(self, thread_id: str) -> threading.Lock:
                 self._thread_locks[thread_id] = lock
             return lock
 
-    def _record_to_connection(self, thread_id: str, record: SandboxRecord) -> SandboxConnection:
+    def _record_to_connection(self, thread_id: str, user_id: str, record: SandboxRecord) -> SandboxConnection:
         connection = SandboxConnection(
             thread_id=thread_id,
+            user_id=user_id,
             sandbox_id=record.sandbox_id,
             sandbox_url=record.sandbox_url,
         )
@@ -73,7 +75,7 @@ def _touch_if_needed(self, connection: SandboxConnection) -> bool:
         self._last_touch_at[connection.thread_id] = time.time()
         return is_alive
 
-    def acquire(self, thread_id: str) -> str:
+    def acquire(self, thread_id: str, *, user_id: str) -> str:
         lock = self._thread_lock(thread_id)
         with lock:
             current = self._connections.get(thread_id)
@@ -91,14 +93,14 @@ def acquire(self, thread_id: str) -> str:
             record = self._client.discover(sandbox_id)
             if record is None:
                 logger.info(f"Creating sandbox {sandbox_id} for thread {thread_id}")
-                record = self._client.create(sandbox_id, thread_id)
+                record = self._client.create(sandbox_id, thread_id, user_id)
             else:
                 logger.info(f"Reusing sandbox {sandbox_id} for thread {thread_id}")
 
-            connection = self._record_to_connection(thread_id, record)
+            connection = self._record_to_connection(thread_id, user_id, record)
             return connection.sandbox_id
 
-    def get(self, thread_id: str, *, create_if_missing: bool = False) -> SandboxConnection | None:
+    def get(self, thread_id: str, *, user_id: str, create_if_missing: bool = False) -> SandboxConnection | None:
         lock = self._thread_lock(thread_id)
         with lock:
             current = self._connections.get(thread_id)
@@ -111,19 +113,19 @@ def get(self, thread_id: str, *, create_if_missing: bool = False) -> SandboxConn
                 except Exception as exc:  # noqa: BLE001
                     logger.warning(f"Failed to touch sandbox {current.sandbox_id} for thread {thread_id}: {exc}")
                     return current
-
-            current = self._connections.get(thread_id)
-            if current:
-                return current
+                if current.user_id == user_id:
+                    return current
+                self._connections.pop(thread_id, None)
+                self._last_touch_at.pop(thread_id, None)
 
             sandbox_id = sandbox_id_for_thread(thread_id)
             record = self._client.discover(sandbox_id)
             if record is None:
                 if not create_if_missing:
                     return None
-                record = self._client.create(sandbox_id, thread_id)
+                record = self._client.create(sandbox_id, thread_id, user_id)
 
-            return self._record_to_connection(thread_id, record)
+            return self._record_to_connection(thread_id, user_id, record)
 
     def shutdown(self) -> None:
         with self._lock:

backend/package/yuxi/agents/backends/sandbox/provisioner_client.py

@@ -29,11 +29,11 @@ def health(self) -> bool:
         response = self._request("GET", "/health")
         return response.status_code == 200
 
-    def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord:
+    def create(self, sandbox_id: str, thread_id: str, user_id: str) -> SandboxRecord:
         response = self._request(
             "POST",
             "/api/sandboxes",
-            json={"sandbox_id": sandbox_id, "thread_id": thread_id},
+            json={"sandbox_id": sandbox_id, "thread_id": thread_id, "user_id": user_id},
         )
         if response.status_code >= 400:
             raise RuntimeError(f"failed to create sandbox {sandbox_id}: {response.status_code} {response.text}")

backend/package/yuxi/agents/toolkits/buildin/tools.py

@@ -83,8 +83,11 @@ def _normalize_presented_artifact_path(filepath: str, runtime: ToolRuntime) -> s
     thread_id = getattr(runtime_context, "thread_id", None)
     if not thread_id:
         raise ValueError("当前运行时缺少 thread_id")
+    user_id = getattr(runtime_context, "user_id", None)
+    if not user_id:
+        raise ValueError("当前运行时缺少 user_id")
 
-    ensure_thread_dirs(thread_id)
+    ensure_thread_dirs(thread_id, str(user_id))
     outputs_dir = sandbox_outputs_dir(thread_id).resolve()
     normalized_input = str(filepath or "").strip()
     if not normalized_input:
@@ -93,7 +96,7 @@ def _normalize_presented_artifact_path(filepath: str, runtime: ToolRuntime) -> s
     stripped = normalized_input.lstrip("/")
     virtual_prefix = VIRTUAL_PATH_PREFIX.lstrip("/")
     if stripped == virtual_prefix or stripped.startswith(f"{virtual_prefix}/"):
-        actual_path = resolve_virtual_path(thread_id, normalized_input)
+        actual_path = resolve_virtual_path(thread_id, normalized_input, user_id=str(user_id))
     else:
         actual_path = Path(normalized_input).expanduser().resolve()
 

backend/package/yuxi/services/conversation_service.py

@@ -209,12 +209,13 @@ def serialize_attachment(record: dict) -> dict:
 async def _materialize_attachment_files(
     *,
     thread_id: str,
+    user_id: str,
     upload: UploadFile,
     file_name: str,
     file_content: bytes,
 ) -> dict:
     """将原始附件与可选 markdown 副本落盘到线程 user-data。"""
-    ensure_thread_dirs(thread_id)
+    ensure_thread_dirs(thread_id, user_id)
 
     upload_virtual_path = _make_upload_virtual_path(file_name)
     uploads_dir = sandbox_uploads_dir(thread_id)
@@ -384,6 +385,7 @@ async def upload_thread_attachment_view(
         raise HTTPException(status_code=400, detail=f"附件过大，当前仅支持 {max_size_mb} MB 以内的文件")
     materialized = await _materialize_attachment_files(
         thread_id=thread_id,
+        user_id=str(conversation.user_id),
         upload=file,
         file_name=file_name,
         file_content=file_content,

backend/package/yuxi/services/filesystem_service.py

@@ -67,7 +67,7 @@ async def _resolve_filesystem_state(
     runtime_context.user_id = str(user.id)
     await resolve_visible_knowledge_bases_for_context(runtime_context)
 
-    sandbox_backend = ProvisionerSandboxBackend(thread_id=thread_id)
+    sandbox_backend = ProvisionerSandboxBackend(thread_id=thread_id, user_id=str(user.id))
     return conversation, runtime_context, sandbox_backend