Fix: private dataset authorization bypass in shared dataset access checks

agent/canvas.py

@@ -379,8 +379,10 @@ async def run(self, **kwargs):
         self.message_id = get_uuid()
         created_at = int(time.time())
         self.add_user_input(kwargs.get("query"))
+        path_set = set(self.path)
         for k, cpn in self.components.items():
-            self.components[k]["obj"].reset(True)
+            if k in path_set:
+                self.components[k]["obj"].reset(True)
 
         if kwargs.get("webhook_payload"):
             for k, cpn in self.components.items():

agent/sandbox/client.py

@@ -23,11 +23,12 @@
 
 import json
 import logging
+import os
 from typing import Dict, Any, Optional
 
 from api.db.services.system_settings_service import SystemSettingsService
 from agent.sandbox.providers import ProviderManager
-from agent.sandbox.providers.base import ExecutionResult
+from agent.sandbox.providers.base import ExecutionResult, SandboxProviderConfigError
 
 logger = logging.getLogger(__name__)
 
@@ -59,50 +60,36 @@ def _load_provider_from_settings() -> None:
     """
     Load sandbox provider from system settings and configure the provider manager.
 
-    This function reads the system settings to determine which provider is active
-    and initializes it with the appropriate configuration.
+    This function resolves the active provider type, then loads configuration
+    from system settings with environment overrides for that provider.
     """
     global _provider_manager
 
     if _provider_manager is None:
         return
 
     try:
-        # Get active provider type
-        provider_type_settings = SystemSettingsService.get_by_name("sandbox.provider_type")
-        if not provider_type_settings:
-            raise RuntimeError(
-                "Sandbox provider type not configured. Please set 'sandbox.provider_type' in system settings."
-            )
-        provider_type = provider_type_settings[0].value
-
-        # Get provider configuration
-        provider_config_settings = SystemSettingsService.get_by_name(f"sandbox.{provider_type}")
-
-        if not provider_config_settings:
-            logger.warning(f"No configuration found for provider: {provider_type}")
-            config = {}
-        else:
-            try:
-                config = json.loads(provider_config_settings[0].value)
-            except json.JSONDecodeError as e:
-                logger.error(f"Failed to parse sandbox config for {provider_type}: {e}")
-                config = {}
+        provider_type, provider_type_from_env = _resolve_provider_type()
+        config = _load_provider_config(provider_type)
 
         # Import and instantiate the provider
         from agent.sandbox.providers import (
             SelfManagedProvider,
             AliyunCodeInterpreterProvider,
             E2BProvider,
+            LocalProvider,
         )
 
         provider_classes = {
             "self_managed": SelfManagedProvider,
             "aliyun_codeinterpreter": AliyunCodeInterpreterProvider,
             "e2b": E2BProvider,
+            "local": LocalProvider,
         }
 
         if provider_type not in provider_classes:
+            if provider_type_from_env:
+                raise SandboxProviderConfigError(f"Unknown sandbox provider type: {provider_type}")
             logger.error(f"Unknown provider type: {provider_type}")
             return
 
@@ -111,19 +98,97 @@ def _load_provider_from_settings() -> None:
 
         # Initialize the provider
         if not provider.initialize(config):
-            logger.error(f"Failed to initialize sandbox provider: {provider_type}. Config keys: {list(config.keys())}")
+            message = f"Failed to initialize sandbox provider: {provider_type}. Config keys: {list(config.keys())}"
+            if provider_type == "local" or provider_type_from_env:
+                raise SandboxProviderConfigError(message)
+            logger.error(message)
             return
 
         # Set the active provider
         _provider_manager.set_provider(provider_type, provider)
         logger.info(f"Sandbox provider '{provider_type}' initialized successfully")
 
+    except SandboxProviderConfigError:
+        raise
     except Exception as e:
         logger.error(f"Failed to load sandbox provider from settings: {e}")
         import traceback
         traceback.print_exc()
 
 
+def _load_provider_config_from_settings(provider_type: str) -> Dict[str, Any]:
+    provider_config_settings = SystemSettingsService.get_by_name(f"sandbox.{provider_type}")
+    if not provider_config_settings:
+        logger.warning(f"No configuration found for provider: {provider_type}")
+        return {}
+
+    try:
+        return json.loads(provider_config_settings[0].value)
+    except json.JSONDecodeError as e:
+        logger.error(f"Failed to parse sandbox config for {provider_type}: {e}")
+        return {}
+
+
+def _resolve_provider_type() -> tuple[str, bool]:
+    provider_type = os.environ.get("SANDBOX_PROVIDER_TYPE", "").strip()
+    if provider_type:
+        return provider_type, True
+
+    provider_type_settings = SystemSettingsService.get_by_name("sandbox.provider_type")
+    if not provider_type_settings:
+        raise RuntimeError(
+            "Sandbox provider type not configured. Please set 'sandbox.provider_type' in system settings."
+        )
+    return provider_type_settings[0].value, False
+
+
+def _load_provider_config(provider_type: str) -> Dict[str, Any]:
+    config = _load_provider_config_from_settings(provider_type)
+    env_config = _load_provider_config_from_env(provider_type)
+    if env_config:
+        config.update(env_config)
+    return config
+
+
+def _load_provider_config_from_env(provider_type: str) -> Dict[str, Any]:
+    if provider_type == "local":
+        return _load_local_provider_config_from_env()
+    if provider_type == "self_managed":
+        return _load_self_managed_provider_config_from_env()
+    return {}
+
+
+def _load_local_provider_config_from_env() -> Dict[str, Any]:
+    env_to_config = {
+        "SANDBOX_LOCAL_PYTHON_BIN": "python_bin",
+        "SANDBOX_LOCAL_NODE_BIN": "node_bin",
+        "SANDBOX_LOCAL_WORK_DIR": "work_dir",
+        "SANDBOX_LOCAL_TIMEOUT": "timeout",
+        "SANDBOX_LOCAL_MAX_MEMORY_MB": "max_memory_mb",
+        "SANDBOX_LOCAL_MAX_OUTPUT_BYTES": "max_output_bytes",
+        "SANDBOX_LOCAL_MAX_ARTIFACTS": "max_artifacts",
+        "SANDBOX_LOCAL_MAX_ARTIFACT_BYTES": "max_artifact_bytes",
+    }
+    config = {}
+    for env_name, config_name in env_to_config.items():
+        if env_name in os.environ:
+            config[config_name] = os.environ[env_name]
+    return config
+
+
+def _load_self_managed_provider_config_from_env() -> Dict[str, Any]:
+    host = os.environ.get("SANDBOX_HOST", "").strip()
+    port = os.environ.get("SANDBOX_EXECUTOR_MANAGER_PORT", "").strip()
+    pool_size = os.environ.get("SANDBOX_EXECUTOR_MANAGER_POOL_SIZE", "").strip()
+
+    config = {}
+    if host:
+        config["endpoint"] = f"http://{host}:{port or '9385'}"
+    if pool_size:
+        config["pool_size"] = pool_size
+    return config
+
+
 def reload_provider() -> None:
     """
     Reload the sandbox provider from system settings.

agent/sandbox/providers/__init__.py

@@ -24,20 +24,24 @@
 - aliyun_codeinterpreter.py: Aliyun Code Interpreter provider implementation
   Official Documentation: https://help.aliyun.com/zh/functioncompute/fc/sandbox-sandbox-code-interepreter
 - e2b.py: E2B provider implementation
+- local.py: Local process provider implementation
 """
 
-from .base import SandboxProvider, SandboxInstance, ExecutionResult
+from .base import SandboxProvider, SandboxInstance, ExecutionResult, SandboxProviderConfigError
 from .manager import ProviderManager
 from .self_managed import SelfManagedProvider
 from .aliyun_codeinterpreter import AliyunCodeInterpreterProvider
 from .e2b import E2BProvider
+from .local import LocalProvider
 
 __all__ = [
     "SandboxProvider",
     "SandboxInstance",
     "ExecutionResult",
+    "SandboxProviderConfigError",
     "ProviderManager",
     "SelfManagedProvider",
     "AliyunCodeInterpreterProvider",
     "E2BProvider",
+    "LocalProvider",
 ]

agent/sandbox/providers/aliyun_codeinterpreter.py

@@ -30,7 +30,6 @@
 import logging
 import os
 import time
-import base64
 import json
 from typing import Dict, Any, List, Optional
 from datetime import datetime, timezone
@@ -39,10 +38,10 @@
 from agentrun.utils.config import Config
 from agentrun.utils.exception import ServerError
 
+from agent.sandbox.result_protocol import build_javascript_wrapper, build_python_wrapper, extract_structured_result
 from .base import SandboxProvider, SandboxInstance, ExecutionResult
 
 logger = logging.getLogger(__name__)
-RESULT_MARKER_PREFIX = "__RAGFLOW_RESULT__:"
 
 
 class AliyunCodeInterpreterProvider(SandboxProvider):
@@ -234,9 +233,9 @@ def execute_code(self, instance_id: str, code: str, language: str, timeout: int
             # Matches self_managed provider behavior: call main(**arguments)
             args_json = json.dumps(arguments or {})
             wrapped_code = (
-                self._build_python_wrapper(code, args_json)
+                build_python_wrapper(code, args_json)
                 if normalized_lang == "python"
-                else self._build_javascript_wrapper(code, args_json)
+                else build_javascript_wrapper(code, args_json)
             )
             logger.debug(f"Aliyun Code Interpreter: Wrapped code (first 200 chars): {wrapped_code[:200]}")
 
@@ -284,7 +283,7 @@ def execute_code(self, instance_id: str, code: str, language: str, timeout: int
 
             stdout = "\n".join(stdout_parts)
             stderr = "\n".join(stderr_parts)
-            stdout, structured_result = self._extract_structured_result(stdout)
+            stdout, structured_result = extract_structured_result(stdout)
 
             logger.info(f"Aliyun Code Interpreter: stdout length={len(stdout)}, stderr length={len(stderr)}, exit_code={exit_code}")
             if stdout:
@@ -364,71 +363,6 @@ def health_check(self) -> bool:
             # If we get any response (even an error), the service is reachable
             return "connection" not in str(e).lower()
 
-    @staticmethod
-    def _build_python_wrapper(code: str, args_json: str) -> str:
-        marker = RESULT_MARKER_PREFIX
-        return f'''{code}
-
-if __name__ == "__main__":
-    import base64
-    import json
-
-    result = main(**{args_json})
-    payload = json.dumps({{"present": True, "value": result, "type": "json"}}, ensure_ascii=False, separators=(",", ":"))
-    print("{marker}" + base64.b64encode(payload.encode("utf-8")).decode("ascii"))
-'''
-
-    @staticmethod
-    def _build_javascript_wrapper(code: str, args_json: str) -> str:
-        marker = RESULT_MARKER_PREFIX
-        return f'''{code}
-
-const __ragflowArgs = {args_json};
-
-(async () => {{
-  try {{
-    const output = await Promise.resolve(main(__ragflowArgs));
-    if (typeof output === 'undefined') {{
-      throw new Error('main() must return a value. Use null for an empty result.');
-    }}
-    const payload = JSON.stringify({{ present: true, value: output, type: 'json' }});
-    if (typeof payload === 'undefined') {{
-      throw new Error('main() returned a non-JSON-serializable value.');
-    }}
-    console.log('{marker}' + Buffer.from(payload, 'utf8').toString('base64'));
-  }} catch (err) {{
-    console.error(err instanceof Error ? err.stack || err.message : String(err));
-  }}
-}})();
-'''
-
-    @staticmethod
-    def _extract_structured_result(stdout: str) -> tuple[str, Dict[str, Any]]:
-        if not stdout:
-            return "", {}
-
-        cleaned_lines: list[str] = []
-        structured_result: Dict[str, Any] = {}
-
-        for line in str(stdout).splitlines():
-            if line.startswith(RESULT_MARKER_PREFIX):
-                payload_b64 = line[len(RESULT_MARKER_PREFIX) :].strip()
-                if not payload_b64:
-                    continue
-                try:
-                    payload = base64.b64decode(payload_b64).decode("utf-8")
-                    structured_result = json.loads(payload)
-                except Exception as exc:
-                    logger.warning(f"Aliyun Code Interpreter: failed to decode structured result marker: {exc}")
-                    cleaned_lines.append(line)
-                continue
-            cleaned_lines.append(line)
-
-        cleaned_stdout = "\n".join(cleaned_lines)
-        if stdout.endswith("\n") and cleaned_stdout and not cleaned_stdout.endswith("\n"):
-            cleaned_stdout += "\n"
-        return cleaned_stdout, structured_result
-
     def get_supported_languages(self) -> List[str]:
         """
         Get list of supported programming languages.

agent/sandbox/providers/base.py

@@ -26,6 +26,10 @@
 from typing import Dict, Any, Optional, List
 
 
+class SandboxProviderConfigError(Exception):
+    """Raised when the selected provider is explicitly configured but unusable."""
+
+
 @dataclass
 class SandboxInstance:
     """Represents a sandbox execution instance"""
@@ -209,4 +213,4 @@ def validate_config(self, config: Dict[str, Any]) -> tuple[bool, Optional[str]]:
             >>>     return True, None
         """
         # Default implementation: no custom validation
-        return True, None
+        return True, None