diff --git a/README.md b/README.md index 7139b1243..5fad72e34 100644 --- a/README.md +++ b/README.md @@ -591,6 +591,12 @@ DeerFlow supports multiple sandbox execution modes: - **Docker Execution** (runs sandbox code in isolated Docker containers) - **Docker Execution with Kubernetes** (runs sandbox code in Kubernetes pods via provisioner service) +Sandbox references in conversation state are server-owned. External run and +thread-state APIs reject caller-supplied `sandbox` values; when restoring a +checkpoint, the runtime resolves the reference against the authenticated user +and thread before a tool can reuse it. A missing runtime thread ID raises an +error even when the referenced sandbox is cached. + When host Bash is enabled for Local Execution, DeerFlow starts OS detection with `uname -s`, then uses `sw_vers` on Darwin. On Linux, it reads host system files such as `/etc/os-release` only when the active sandbox policy permits it. Host filesystem path checks still apply; after a blocked path, the agent is directed to use a permitted command-only probe or virtual path instead of repeating the rejected command. For Docker development, service startup follows `config.yaml` sandbox mode. In Local/Docker modes, `provisioner` is not started. diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 38d759127..6dfb70961 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -451,7 +451,8 @@ def _normalize_input_messages( def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, Any]: """Validate and sanitize caller-supplied state values before checkpointing. - The ``messages`` channel is canonicalized to a list of ``BaseMessage`` + The server-owned ``sandbox`` channel is rejected. The ``messages`` channel + is canonicalized to a list of ``BaseMessage`` objects, rejects external system/developer roles with HTTP 400, and strips server-owned metadata. Other channels keep their existing shapes while forged metadata and delegation verdicts are removed. ``normalize_input`` @@ -462,6 +463,12 @@ def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, An transform trails, or privileged message roles. Every channel is walked because middleware-contributed channels can also carry message-like values. """ + if "sandbox" in values: + raise HTTPException( + status_code=400, + detail="External sandbox state is not allowed", + ) + stripped: dict[str, Any] = {} for channel, value in values.items(): if channel == "messages" and value is not None: @@ -491,6 +498,10 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool of bubbling up as a 500. The gateway is a system boundary, so per-entry validation errors are the right shape for clients to retry against. + The ``sandbox`` channel is also server-owned. External callers cannot select + a provider resource by id; trusted internal run admission may carry the + server's own restored value. + ``original_user_content``, dynamic-context reminder markers, the transient view-image context marker, the execution-only knowledge-scope marker, tool receipts, delegated receipt metadata/verdicts, and ``untrusted_input`` are @@ -513,6 +524,11 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool """ if raw_input is None: return {} + if not trusted_internal and "sandbox" in raw_input: + raise HTTPException( + status_code=400, + detail="External sandbox state is not allowed", + ) result = raw_input messages = raw_input.get("messages") if messages is not None: diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index 3fd103f24..9620b799b 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -2398,6 +2398,25 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._last_activity[sandbox_id] = time.time() return sandbox + def get_scoped( + self, + sandbox_id: str, + *, + thread_id: str, + user_id: str, + ) -> Sandbox | None: + """Return a cached client only for its recorded user/thread identity.""" + key = self._thread_key(thread_id, user_id) + with self._lock: + if self._thread_sandboxes.get(key) != sandbox_id: + return None + if self._active_sandbox_identity.get(sandbox_id) != key: + return None + sandbox = self._sandboxes.get(sandbox_id) + if sandbox is not None: + self._last_activity[sandbox_id] = time.time() + return sandbox + def release(self, sandbox_id: str) -> None: """Release a sandbox from active use. diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index f8027bb00..3d51f9899 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -1,5 +1,7 @@ ### Sandbox System (`packages/harness/deerflow/sandbox/`) +Sandbox restore requires a thread ID, even forks. + **Network approval policy**: Sync/async `SandboxMiddleware` wrappers use `resolve_run_interaction_policy()` to gate lead-run network approval cards. Explicit `autonomous`, `webhook`, and `scheduled` modes auto-deny pending requests, @@ -9,9 +11,9 @@ interactive runs; never consume events into a human-input card without a human to respond. **Interface**: `Sandbox`: `execute_command(command, env=None)`, additive `execute_command_in_scope(..., scope_id=...)` / `release_command_scope(scope_id)`, `read_file`, `write_file`, `list_dir`, `glob`, `grep`. Scoped hooks default to pass-through without server-side sessions, preserving third-party subclasses. `grep` accepts a text file or directory tree. Per-call `env` injects secrets: `LocalSandbox` merges into the subprocess environment; `AioSandbox` uses fresh `bash.exec(env=...)` sessions. `list_dir`: missing path → `FileNotFoundError`; command/client failure → `OSError`, never `[]` (`ls_tool`: `(empty)`). Remote `glob`/`grep` share it via `sandbox/remote_search.py`: missing root → `FileNotFoundError`, failed search → `OSError`; only a genuine no-match returns `[]`. The parser takes the command's `limit` and reports `truncated` when output passed it, which providers return after Python-side filtering; tools call an empty truncated result incomplete. Remote `grep(glob=...)` scopes like `glob()` (root-relative `path_matches`), never by basename alone. Remotes use `sandbox/remote_list_dir.py`: capture `find`'s status, not `| head`'s (`sh -lc` lacks `pipefail`); missing binary → `OSError`; truncation SIGPIPE → success. -**Provider Pattern**: `SandboxProvider` exposes `acquire`, `acquire_async`, `get`, `release`. Async agent/tool paths use async hooks to keep Docker creation, discovery, cross-process locking, readiness polling, and release off-loop. Set `supports_agent_skill_isolation=True` only when the whole tool surface enforces explicit lead Agent policy: bind mounts use prepared thread roots; upload providers implement `sync_agent_skills`. Host-backed providers report false if an enabled shell bypasses path mappings. Under explicit policy, middleware rejects unsupported providers before acquire. +**Provider Pattern**: `SandboxProvider`: `acquire`, `acquire_async`, `get`, `get_scoped`, `release`. `get_scoped` checks the identity-scoped cache without blocking; a miss uses canonical `acquire(user_id, thread_id)`, never a checkpoint id. AIO reads active identity maps. Async hooks keep Docker creation/discovery, cross-process locking, readiness checks, and release off-loop. Set `supports_agent_skill_isolation=True` only if every tool enforces explicit lead Agent policy: prepared thread-root bind mounts plus upload `sync_agent_skills`. Host-backed providers report false when enabled shells bypass path mappings; middleware rejects them before explicit-policy acquire. **Shared components** (RFC #4741): remote IDs use `derive_sandbox_scope_token` (`sandbox/identity.py`); preserve its keyword-only SHA-256/16-hex contract to avoid orphaning containers. `AcquireSerializer` (`sandbox/acquire_serialization.py`) serializes selected acquire/release transitions with a bounded, refcounted per-key `threading.Lock` table and dedicated bounded executor (no event-loop/default-executor blocking). Workers own cancellation cleanup without waiting for cancelled tasks to resume; provider `shutdown()`/`reset()` calls idempotent `close()`. Keys: AIO `(user_id, thread_id)`, E2B `(user_id, thread_id, skills_root)`, BoxLite/Tenki/OpenSandbox derived id. Random-UUID `thread_id=None` acquires bypass serialization. -**Execution leases** (`sandbox/lease.py`, #5128): cross-instance ownership decides which Gateway may reap a container; process-local `SandboxLeaseManager` tracks concurrent lead, subagent, Gateway-request, and channel-upload users of one client. Runs get ephemeral owners, persisted sandboxes are retained idempotently, and the last holder performs any pending `SandboxProvider.release`. Outer lifecycle fences repeat idempotent release after their complete graph/tool/request batch drains; per-tool terminal `Command` wrappers never release because sibling handlers may still run. Fork-restored children and upload syncs use non-releasing holders: they fence the client and own scope cleanup without themselves requesting a park; an earlier normal-owner request waits for them, and a missing fork client is replaced by a normal owner. Persisted lookup plus retention is serialized per `(user_id, thread_id)`; stale bindings fall through to acquire, and a post-acquire lookup miss rolls back before raising. Provider I/O does not hold the metadata lock. Repeated cancellation cannot interrupt acquire/rollback/release reconciliation or let a `to_thread` sandbox operation outlive its enclosing execution/request holder; failures are logged without replacing the original cancellation. Lease/scope context IDs are server-owned: Gateway and worker scrub caller values; only the internal subagent path assigns a task ID. Managers are registered by provider object identity, not hash/equality, so unhashable custom providers remain valid. Subagent owners also serve as `sandbox_command_scope_id`: AIO gives each scope one ordered persistent shell session, replaces it after `ErrorObservation` or a missing-session 404, and cleans it on lease release. Registry identity is revalidated after every scope-lock wait, preventing queued commands from resurrecting released sessions. Env-bearing commands use fresh `bash.exec` sessions so secrets do not persist. +**Execution leases** (`sandbox/lease.py`, #5128): cross-instance ownership selects which Gateway may reap a container; process-local `SandboxLeaseManager` tracks a client's concurrent lead, subagent, Gateway-request, and channel-upload users. Runs have ephemeral owners. Restore persisted sandboxes only after `get_scoped(user_id, thread_id)` succeeds; otherwise canonical `acquire` replaces the id. Last holder performs pending `SandboxProvider.release`. External run/state boundaries reject server-owned `sandbox`. Outer lifecycle fences repeat idempotent release after each graph/tool/request batch; per-tool terminal `Command` wrappers never release while siblings run. Fork-restored children and upload syncs hold without requesting release: server-created forks may borrow the parent client; ordinary checkpoint ids stay identity-scoped. Serialize persisted lookup and retention by `(user_id, thread_id)`; stale or mismatched bindings reacquire, and a post-acquire miss rolls back then raises. Provider I/O runs outside the metadata lock. Repeated cancellation cannot interrupt acquire/rollback/release reconciliation or let `to_thread` work outlive its execution/request holder; failures are logged without replacing the original cancellation. Lease/scope context IDs are server-owned: Gateway/worker scrub caller values; only internal subagents assign task IDs. Managers key by provider object identity (not hash/equality), so unhashable custom providers work. Subagent owners double as `sandbox_command_scope_id`: AIO gives each scope an ordered persistent shell, replaces it after `ErrorObservation` or a missing-session 404, and cleans it on lease release. Revalidate registry identity after each scope-lock wait so queued commands cannot resurrect released sessions. Env-bearing commands use fresh `bash.exec` sessions so secrets do not persist. **Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox-backed tool call passes through the gate in `deerflow/authz/sandbox_authz.py` - a binary `authorize(principal, "sandbox", "execute", target="*")` check before either reusing a persisted sandbox id or calling `provider.acquire`. Rechecking reuse is required because authorization config and user roles can change while the sandbox remains cached. Sync tool invocations call `authorize_sandbox_execution`; async tool invocations await `authorize_sandbox_execution_async` exactly once. A task-local `ContextVar` scopes that single decision across the complete composed tool invocation, including `ReadBeforeWriteMiddleware`'s pre-write inspection, tool body, and post-read mark; the value is copied into `asyncio.to_thread` workers. Authorization denial is converted to the normal error `ToolMessage` at the composed middleware boundary and is explicitly excluded from the gate's generic fail-open handlers. Async config loading and provider class discovery/import are offloaded before `aauthorize()` so reused sandbox calls do not hash config files or import custom modules on the event loop; provider construction remains on the running event loop because async providers may initialize loop-affine clients. The gate lives at the single tool initialization entry point (`ensure_sandbox_initialized` / `ensure_sandbox_initialized_async` in `tools.py`), while `SandboxMiddleware.before_agent` / `abefore_agent` apply the matching sync/async check to eager acquisition. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of ordinary tool execution as a friendly error `ToolMessage` ("sandbox execution is not permitted for your role") - the eager path catches it and skips acquisition instead, deferring the deny to the first sandbox-touching tool call so both paths share the same semantics. Provider errors (authorization calls and provider resolution) follow `authorization.fail_closed` / `fail_open`; no readable `config.yaml` or `authorization.enabled: false` makes the gate a no-op (`safe_app_config` tolerates missing config). Gateway upload/artifact sync calls `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates, returns a request lease, and skips sync on deny while preserving the primary operation. Callers release after their last sandbox operation; artifacts request normal parking, uploads do not. Tests: `tests/test_sandbox_authorization.py` and `tests/blocking_io/test_sandbox_authorization.py`. **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved. **Implementations**: diff --git a/backend/packages/harness/deerflow/sandbox/lease.py b/backend/packages/harness/deerflow/sandbox/lease.py index b0878cbe8..15170def1 100644 --- a/backend/packages/harness/deerflow/sandbox/lease.py +++ b/backend/packages/harness/deerflow/sandbox/lease.py @@ -436,12 +436,15 @@ class SandboxLeaseManager: user_id: str, release_on_last: bool = True, acquire_release_on_last: bool = True, + allow_unscoped_borrow: bool = False, ) -> str: - """Atomically retain a live persisted sandbox or acquire a replacement. + """Atomically restore a scoped sandbox or acquire a replacement. - A fork-restored live client is borrowed with ``release_on_last=False``; - if that persisted client is gone, its freshly acquired replacement is - owned normally unless ``acquire_release_on_last`` is also disabled. + Ordinary checkpoint ids must match ``user_id`` and ``thread_id``. + Server-created fork wrappers may set ``allow_unscoped_borrow`` with + ``release_on_last=False`` to share a parent's live client. If that + client is gone, the replacement is owned normally unless + ``acquire_release_on_last`` is also disabled. """ key = self._thread_key(thread_id, user_id) with self._serializer.hold(key): @@ -453,7 +456,16 @@ class SandboxLeaseManager: if existing_sandbox_id is not None: return existing_sandbox_id - if self._provider.get(sandbox_id) is not None: + sandbox = ( + self._provider.get(sandbox_id) + if allow_unscoped_borrow + else self._provider.get_scoped( + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) + ) + if sandbox is not None: with self._metadata_lock: previous, release_previous = self._bind_locked( owner_id, @@ -508,6 +520,7 @@ class SandboxLeaseManager: user_id: str, release_on_last: bool = True, acquire_release_on_last: bool = True, + allow_unscoped_borrow: bool = False, ) -> str: """Async atomic retain-or-replace transition for a persisted sandbox.""" key = self._thread_key(thread_id, user_id) @@ -520,7 +533,16 @@ class SandboxLeaseManager: if existing_sandbox_id is not None: return existing_sandbox_id - if self._provider.get(sandbox_id) is not None: + sandbox = ( + self._provider.get(sandbox_id) + if allow_unscoped_borrow + else self._provider.get_scoped( + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) + ) + if sandbox is not None: with self._metadata_lock: previous, release_previous = self._bind_locked( owner_id, diff --git a/backend/packages/harness/deerflow/sandbox/middleware.py b/backend/packages/harness/deerflow/sandbox/middleware.py index 0e29de3be..7bb11c048 100644 --- a/backend/packages/harness/deerflow/sandbox/middleware.py +++ b/backend/packages/harness/deerflow/sandbox/middleware.py @@ -178,7 +178,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): sandbox_id = sandbox.get("sandbox_id") if isinstance(sandbox_id, str): provider = get_sandbox_provider() - get_sandbox_lease_manager(provider).retain( + sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire( owner_id, sandbox_id, thread_id=thread_id, @@ -203,7 +203,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): sandbox_id = sandbox.get("sandbox_id") if isinstance(sandbox_id, str): provider = get_sandbox_provider() - await get_sandbox_lease_manager(provider).retain_async( + sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async( owner_id, sandbox_id, thread_id=thread_id, @@ -306,8 +306,11 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): user_id=user_id, owner_id=owner_id, ) - if retained_id is not None and runtime.context is not None: - runtime.context["sandbox_id"] = retained_id + if retained_id is not None: + if runtime.context is not None: + runtime.context["sandbox_id"] = retained_id + if retained_id != existing_sandbox_id: + return {"sandbox": Overwrite({"sandbox_id": retained_id})} return super().before_agent(state, runtime) def _apply_network_policy_response(self, state: SandboxMiddlewareState, runtime: Runtime) -> None: @@ -416,8 +419,11 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): user_id=user_id, owner_id=owner_id, ) - if retained_id is not None and runtime.context is not None: - runtime.context["sandbox_id"] = retained_id + if retained_id is not None: + if runtime.context is not None: + runtime.context["sandbox_id"] = retained_id + if retained_id != existing_sandbox_id: + return {"sandbox": Overwrite({"sandbox_id": retained_id})} return await super().abefore_agent(state, runtime) @override @@ -508,7 +514,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): return sandbox_id if isinstance(sandbox_id, str) else None @staticmethod - def _attach_sandbox_update(result: ToolMessage | Command, sandbox_id: str) -> ToolMessage | Command: + def _attach_sandbox_update( + result: ToolMessage | Command, + sandbox_id: str, + *, + overwrite: bool = False, + ) -> ToolMessage | Command: """Wrap or merge ``result`` so that ``sandbox.sandbox_id`` is persisted. - ``ToolMessage`` -> ``Command(update={"sandbox": ..., "messages": [msg]})`` @@ -517,7 +528,10 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): - ``Command`` with non-dict / None update -> leave it untouched to avoid silent data loss on unknown update shapes. """ - sandbox_update = {"sandbox": {"sandbox_id": sandbox_id}} + sandbox_value: object = {"sandbox_id": sandbox_id} + if overwrite: + sandbox_value = Overwrite(sandbox_value) + sandbox_update = {"sandbox": sandbox_value} if isinstance(result, ToolMessage): return Command(update={**sandbox_update, "messages": [result]}) @@ -545,8 +559,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): prev_sandbox_id = self._read_sandbox_id_from_request(request) result = handler(request) curr_sandbox_id = self._read_sandbox_id_from_request(request) - if prev_sandbox_id is None and curr_sandbox_id is not None: - result = self._attach_sandbox_update(result, curr_sandbox_id) + if curr_sandbox_id is not None and curr_sandbox_id != prev_sandbox_id: + result = self._attach_sandbox_update( + result, + curr_sandbox_id, + overwrite=prev_sandbox_id is not None, + ) return self._maybe_request_network_approval(request, result, curr_sandbox_id or prev_sandbox_id) @override @@ -558,8 +576,12 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): prev_sandbox_id = self._read_sandbox_id_from_request(request) result = await handler(request) curr_sandbox_id = self._read_sandbox_id_from_request(request) - if prev_sandbox_id is None and curr_sandbox_id is not None: - result = self._attach_sandbox_update(result, curr_sandbox_id) + if curr_sandbox_id is not None and curr_sandbox_id != prev_sandbox_id: + result = self._attach_sandbox_update( + result, + curr_sandbox_id, + overwrite=prev_sandbox_id is not None, + ) sandbox_id = curr_sandbox_id or prev_sandbox_id if sandbox_id is None: return result @@ -644,5 +666,6 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]): ) update: dict = {"messages": [message], "sandbox": {"sandbox_id": sandbox_id}} if isinstance(result, Command) and isinstance(result.update, dict): - update = {**result.update, **update} + update = {**result.update, "messages": [message]} + update.setdefault("sandbox", {"sandbox_id": sandbox_id}) return Command(update=update, goto=END) diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py index 60fcb53db..399cbd551 100644 --- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py +++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py @@ -81,6 +81,22 @@ class SandboxProvider(ABC): """ pass + def get_scoped( + self, + sandbox_id: str, + *, + thread_id: str, + user_id: str, + ) -> Sandbox | None: + """Return an active sandbox only when it belongs to this identity. + + This hook must remain a non-blocking in-memory lookup. Providers that + do not implement identity-aware lookup fail closed; the caller then + resolves the canonical sandbox through ``acquire``. + """ + del sandbox_id, thread_id, user_id + return None + @abstractmethod def release(self, sandbox_id: str) -> None: """Release a sandbox environment. diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index 63c1c5237..c94d6ff6e 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -1545,22 +1545,44 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox: provider = get_sandbox_provider() owner_id = sandbox_lease_owner(runtime.context) thread_id = _resolve_runtime_thread_id(runtime) - if owner_id is not None and thread_id is not None: - sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire( - owner_id, - sandbox_id, - thread_id=thread_id, - user_id=resolve_runtime_user_id(runtime), - release_on_last=not fork_restored, - ) + user_id = resolve_runtime_user_id(runtime) + if thread_id is not None: + if owner_id is None: + if not fork_restored: + scoped = provider.get_scoped( + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) + if scoped is None: + sandbox_id = provider.acquire(thread_id, user_id=user_id) + elif fork_restored: + # Only the server-created fork wrapper may borrow a sandbox + # from a different thread identity. Ordinary checkpoint ids + # are resolved again from the authenticated user/thread. + sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire( + owner_id, + sandbox_id, + thread_id=thread_id, + user_id=user_id, + release_on_last=False, + allow_unscoped_borrow=True, + ) + else: + sandbox_id = get_sandbox_lease_manager(provider).reuse_or_acquire( + owner_id, + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) if not fork_restored: runtime.state["sandbox"] = {"sandbox_id": sandbox_id} - sandbox = provider.get(sandbox_id) - if sandbox is not None: - if runtime.context is not None: - runtime.context["sandbox_id"] = sandbox_id # Ensure sandbox_id is in context for releasing in after_agent - return sandbox - # Sandbox was released, fall through to acquire new one + sandbox = provider.get(sandbox_id) + if sandbox is not None: + if runtime.context is not None: + runtime.context["sandbox_id"] = sandbox_id # Ensure sandbox_id is in context for releasing in after_agent + return sandbox + # Missing thread scope or released sandbox: use the lazy path below. # Lazy acquisition: get thread_id and acquire sandbox thread_id = _resolve_runtime_thread_id(runtime) @@ -1623,21 +1645,40 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa provider = get_sandbox_provider() owner_id = sandbox_lease_owner(runtime.context) thread_id = _resolve_runtime_thread_id(runtime) - if owner_id is not None and thread_id is not None: - sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async( - owner_id, - sandbox_id, - thread_id=thread_id, - user_id=resolve_runtime_user_id(runtime), - release_on_last=not fork_restored, - ) + user_id = resolve_runtime_user_id(runtime) + if thread_id is not None: + if owner_id is None: + if not fork_restored: + scoped = provider.get_scoped( + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) + if scoped is None: + sandbox_id = await provider.acquire_async(thread_id, user_id=user_id) + elif fork_restored: + sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async( + owner_id, + sandbox_id, + thread_id=thread_id, + user_id=user_id, + release_on_last=False, + allow_unscoped_borrow=True, + ) + else: + sandbox_id = await get_sandbox_lease_manager(provider).reuse_or_acquire_async( + owner_id, + sandbox_id, + thread_id=thread_id, + user_id=user_id, + ) if not fork_restored: runtime.state["sandbox"] = {"sandbox_id": sandbox_id} - sandbox = provider.get(sandbox_id) - if sandbox is not None: - if runtime.context is not None: - runtime.context["sandbox_id"] = sandbox_id - return sandbox + sandbox = provider.get(sandbox_id) + if sandbox is not None: + if runtime.context is not None: + runtime.context["sandbox_id"] = sandbox_id + return sandbox thread_id = _resolve_runtime_thread_id(runtime) if thread_id is None: diff --git a/backend/tests/test_ensure_sandbox_initialized.py b/backend/tests/test_ensure_sandbox_initialized.py index fc53cb977..c8a2d4873 100644 --- a/backend/tests/test_ensure_sandbox_initialized.py +++ b/backend/tests/test_ensure_sandbox_initialized.py @@ -80,6 +80,10 @@ class _RecordingProvider(SandboxProvider): return self.sandbox return None + def get_scoped(self, sandbox_id: str, *, thread_id: str, user_id: str) -> Sandbox | None: + del thread_id, user_id + return self.get(sandbox_id) + def release(self, sandbox_id: str) -> None: self.released.append(sandbox_id) @@ -193,6 +197,7 @@ def test_ensure_sandbox_initialized_unwraps_overwrite_state() -> None: set_sandbox_provider(provider) try: runtime = _make_runtime({"sandbox": Overwrite({"sandbox_id": "parent-sandbox"})}) + runtime.context["thread_id"] = "thread-1" sandbox = ensure_sandbox_initialized(runtime) finally: reset_sandbox_provider() @@ -210,6 +215,7 @@ async def test_ensure_sandbox_initialized_async_unwraps_overwrite_state() -> Non set_sandbox_provider(provider) try: runtime = _make_runtime({"sandbox": Overwrite({"sandbox_id": "parent-sandbox"})}) + runtime.context["thread_id"] = "thread-1" sandbox = await ensure_sandbox_initialized_async(runtime) finally: reset_sandbox_provider() @@ -280,6 +286,7 @@ def test_ensure_sandbox_initialized_plain_state_unchanged() -> None: set_sandbox_provider(provider) try: runtime = _make_runtime({"sandbox": {"sandbox_id": "parent-sandbox"}}) + runtime.context["thread_id"] = "thread-1" sandbox = ensure_sandbox_initialized(runtime) finally: reset_sandbox_provider() @@ -366,6 +373,7 @@ async def test_ensure_sandbox_initialized_async_plain_state_unchanged() -> None: set_sandbox_provider(provider) try: runtime = _make_runtime({"sandbox": {"sandbox_id": "parent-sandbox"}}) + runtime.context["thread_id"] = "thread-1" sandbox = await ensure_sandbox_initialized_async(runtime) finally: reset_sandbox_provider() diff --git a/backend/tests/test_sandbox_leases.py b/backend/tests/test_sandbox_leases.py index a5533fd44..523d39d86 100644 --- a/backend/tests/test_sandbox_leases.py +++ b/backend/tests/test_sandbox_leases.py @@ -71,6 +71,10 @@ class _LeaseProvider(SandboxProvider): def get(self, sandbox_id): return self.sandbox if sandbox_id == self.sandbox.id else None + def get_scoped(self, sandbox_id, *, thread_id, user_id): + del thread_id, user_id + return self.get(sandbox_id) + def release(self, sandbox_id): self.release_calls.append(sandbox_id) @@ -89,6 +93,10 @@ class _UnhashableLeaseProvider(SandboxProvider): def get(self, sandbox_id): return self.sandbox if sandbox_id == self.sandbox.id else None + def get_scoped(self, sandbox_id, *, thread_id, user_id): + del thread_id, user_id + return self.get(sandbox_id) + def release(self, sandbox_id): self.release_calls.append(sandbox_id) diff --git a/backend/tests/test_sandbox_middleware.py b/backend/tests/test_sandbox_middleware.py index 6b61a7742..fe519b2ca 100644 --- a/backend/tests/test_sandbox_middleware.py +++ b/backend/tests/test_sandbox_middleware.py @@ -158,6 +158,10 @@ class _AsyncOnlyProvider(SandboxProvider): return self.sandbox return None + def get_scoped(self, sandbox_id: str, *, thread_id: str, user_id: str) -> Sandbox | None: + del thread_id, user_id + return self.get(sandbox_id) + def release(self, sandbox_id: str) -> None: self.released_ids.append(sandbox_id) return None @@ -335,7 +339,7 @@ def test_explicit_skill_policy_does_not_reuse_checkpointed_sandbox_after_auth_de [ (SandboxMiddleware(lazy_init=True), {}, Runtime(context={"thread_id": "thread-lazy"})), (SandboxMiddleware(lazy_init=False), {}, Runtime(context={})), - (SandboxMiddleware(lazy_init=False), {"sandbox": {"sandbox_id": "existing"}}, Runtime(context={"thread_id": "thread-existing"})), + (SandboxMiddleware(lazy_init=False), {"sandbox": {"sandbox_id": "async-sandbox"}}, Runtime(context={"thread_id": "thread-existing"})), ], ) async def test_abefore_agent_delegates_to_super_when_not_acquiring( @@ -599,6 +603,46 @@ def test_wrap_tool_call_passthrough_when_sandbox_already_in_state() -> None: assert result is original +def test_wrap_tool_call_overwrites_a_repaired_checkpoint_sandbox() -> None: + middleware = SandboxMiddleware() + state: dict = {"sandbox": {"sandbox_id": "foreign"}} + request = _make_tool_call_request(state) + + def handler(req: ToolCallRequest) -> ToolMessage: + req.runtime.state["sandbox"] = {"sandbox_id": "canonical"} + return ToolMessage(content="ok", tool_call_id="call-1", name="bash") + + result = middleware.wrap_tool_call(request, handler) + + assert isinstance(result, Command) + assert isinstance(result.update, dict) + assert isinstance(result.update["sandbox"], Overwrite) + assert result.update["sandbox"].value == {"sandbox_id": "canonical"} + + +def test_network_prompt_preserves_repaired_checkpoint_overwrite() -> None: + provider = _NetworkPolicyProvider() + provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}] + state: dict = {"sandbox": {"sandbox_id": "foreign"}} + request = _make_tool_call_request(state) + + def handler(req: ToolCallRequest) -> ToolMessage: + req.runtime.state["sandbox"] = {"sandbox_id": "canonical"} + return ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash") + + set_sandbox_provider(provider) + try: + result = SandboxMiddleware().wrap_tool_call(request, handler) + finally: + reset_sandbox_provider() + + assert isinstance(result, Command) + assert result.goto == END + assert isinstance(result.update, dict) + assert isinstance(result.update["sandbox"], Overwrite) + assert result.update["sandbox"].value == {"sandbox_id": "canonical"} + + @pytest.mark.parametrize("async_path", [False, True]) @pytest.mark.parametrize( "context", diff --git a/backend/tests/test_sandbox_reference_ownership.py b/backend/tests/test_sandbox_reference_ownership.py new file mode 100644 index 000000000..a3b36b201 --- /dev/null +++ b/backend/tests/test_sandbox_reference_ownership.py @@ -0,0 +1,288 @@ +"""Sandbox references are server-owned and restored from authenticated scope.""" + +from __future__ import annotations + +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from _router_auth_helpers import make_authed_test_app +from fastapi import HTTPException +from fastapi.testclient import TestClient +from langgraph.types import Overwrite + +from app.gateway.routers import runs, thread_runs, threads +from app.gateway.services import normalize_input, strip_server_owned_state_metadata +from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider +from deerflow.sandbox.exceptions import SandboxRuntimeError +from deerflow.sandbox.lease import SANDBOX_LEASE_OWNER_CONTEXT_KEY +from deerflow.sandbox.sandbox import Sandbox +from deerflow.sandbox.sandbox_provider import SandboxProvider, reset_sandbox_provider, set_sandbox_provider +from deerflow.sandbox.search import GrepMatch +from deerflow.sandbox.tools import ensure_sandbox_initialized, ensure_sandbox_initialized_async + +FOREIGN_SANDBOX_ID = "sandbox-user-b-thread-b" +OWN_SANDBOX_ID = "sandbox-user-a-thread-a" + + +class _ScopedSandbox(Sandbox): + def execute_command(self, command, env=None, timeout=None): + return command + + def read_file(self, path, start_line=None, end_line=None): + return self.id + + def download_file(self, path): + return self.id.encode() + + def list_dir(self, path, max_depth=2): + return [] + + def write_file(self, path, content, append=False): + return None + + def glob(self, path, pattern, *, include_dirs=False, max_results=200): + return [], False + + def grep( + self, + path: str, + pattern: str, + *, + glob: str | None = None, + literal: bool = False, + case_sensitive: bool = False, + max_results: int = 100, + ) -> tuple[list[GrepMatch], bool]: + return [], False + + def update_file(self, path, content): + return None + + +class _IdentityScopedProvider(SandboxProvider): + """Models an active foreign sandbox plus canonical current-scope lookup.""" + + def __init__(self) -> None: + self.sandboxes = { + FOREIGN_SANDBOX_ID: _ScopedSandbox(FOREIGN_SANDBOX_ID), + OWN_SANDBOX_ID: _ScopedSandbox(OWN_SANDBOX_ID), + } + self.acquire_calls: list[tuple[str | None, str | None]] = [] + self.get_calls: list[str] = [] + self.scoped_get_calls: list[tuple[str, str, str]] = [] + + def acquire(self, thread_id=None, *, user_id=None): + self.acquire_calls.append((thread_id, user_id)) + assert (user_id, thread_id) == ("user-a", "thread-a") + return OWN_SANDBOX_ID + + async def acquire_async(self, thread_id=None, *, user_id=None): + return self.acquire(thread_id, user_id=user_id) + + def get(self, sandbox_id): + self.get_calls.append(sandbox_id) + return self.sandboxes.get(sandbox_id) + + def get_scoped(self, sandbox_id, *, thread_id, user_id): + self.scoped_get_calls.append((sandbox_id, thread_id, user_id)) + if (sandbox_id, user_id, thread_id) == (OWN_SANDBOX_ID, "user-a", "thread-a"): + return self.sandboxes[sandbox_id] + return None + + def release(self, sandbox_id): + return None + + +def _runtime_with_foreign_checkpoint(): + return SimpleNamespace( + state={"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}, + context={ + SANDBOX_LEASE_OWNER_CONTEXT_KEY: "run-owner-a", + "thread_id": "thread-a", + "user_id": "user-a", + }, + config={"configurable": {"thread_id": "thread-a"}}, + ) + + +@pytest.mark.parametrize( + ("boundary", "payload"), + [ + (normalize_input, {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}), + (strip_server_owned_state_metadata, {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}), + ], +) +def test_external_sandbox_state_is_rejected_at_gateway_boundaries(boundary, payload): + with pytest.raises(HTTPException) as error: + boundary(payload) + + assert error.value.status_code == 400 + assert FOREIGN_SANDBOX_ID not in str(error.value.detail) + + +def test_trusted_internal_run_input_can_restore_server_owned_sandbox_state(): + payload = {"sandbox": {"sandbox_id": OWN_SANDBOX_ID}} + assert normalize_input(payload, trusted_internal=True)["sandbox"] == payload["sandbox"] + + +@pytest.mark.parametrize( + "path", + [ + "/api/threads/sandbox-owner-http/runs", + "/api/threads/sandbox-owner-http/runs/stream", + "/api/threads/sandbox-owner-http/runs/wait", + "/api/runs/stream", + "/api/runs/wait", + ], +) +def test_all_external_run_entrypoints_reject_sandbox_before_worker(monkeypatch, path): + from app.gateway import services + + app = make_authed_test_app() + app.include_router(runs.router) + app.include_router(thread_runs.router) + app.state.stream_bridge = SimpleNamespace() + app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock()) + monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store)) + monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant: object()) + worker = AsyncMock() + monkeypatch.setattr(services, "run_agent", worker) + + with TestClient(app) as client: + response = client.post( + path, + json={ + "input": { + "messages": [{"role": "user", "content": "synthetic"}], + "sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}, + } + }, + ) + + assert response.status_code == 400, response.text + assert FOREIGN_SANDBOX_ID not in response.text + app.state.run_manager.create_or_reject.assert_not_awaited() + worker.assert_not_awaited() + + +def test_external_state_update_rejects_sandbox_before_checkpoint_access(): + app = make_authed_test_app() + app.include_router(threads.router) + + with TestClient(app) as client: + response = client.post( + "/api/threads/sandbox-owner-http/state", + json={"values": {"sandbox": {"sandbox_id": FOREIGN_SANDBOX_ID}}}, + ) + + assert response.status_code == 400, response.text + assert FOREIGN_SANDBOX_ID not in response.text + + +def test_checkpoint_sandbox_is_resolved_against_current_identity_before_sync_use(): + provider = _IdentityScopedProvider() + set_sandbox_provider(provider) + runtime = _runtime_with_foreign_checkpoint() + try: + sandbox = ensure_sandbox_initialized(runtime) + finally: + reset_sandbox_provider() + + assert sandbox.id == OWN_SANDBOX_ID + assert runtime.state["sandbox"] == {"sandbox_id": OWN_SANDBOX_ID} + assert provider.acquire_calls == [("thread-a", "user-a")] + assert provider.scoped_get_calls == [(FOREIGN_SANDBOX_ID, "thread-a", "user-a")] + assert FOREIGN_SANDBOX_ID not in provider.get_calls + + +@pytest.mark.asyncio +async def test_checkpoint_sandbox_is_resolved_against_current_identity_before_async_use(): + provider = _IdentityScopedProvider() + set_sandbox_provider(provider) + runtime = _runtime_with_foreign_checkpoint() + try: + sandbox = await ensure_sandbox_initialized_async(runtime) + finally: + reset_sandbox_provider() + + assert sandbox.id == OWN_SANDBOX_ID + assert runtime.state["sandbox"] == {"sandbox_id": OWN_SANDBOX_ID} + assert provider.acquire_calls == [("thread-a", "user-a")] + assert provider.scoped_get_calls == [(FOREIGN_SANDBOX_ID, "thread-a", "user-a")] + assert FOREIGN_SANDBOX_ID not in provider.get_calls + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_path", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("fork_restored", [False, True], ids=["checkpoint", "fork"]) +@pytest.mark.parametrize("with_lease_owner", [False, True], ids=["unleased", "leased"]) +async def test_checkpoint_sandbox_without_thread_id_fails_closed(async_path, fork_restored, with_lease_owner): + provider = _IdentityScopedProvider() + set_sandbox_provider(provider) + runtime = _runtime_with_foreign_checkpoint() + runtime.context.pop("thread_id") + runtime.config = {} + if not with_lease_owner: + runtime.context.pop(SANDBOX_LEASE_OWNER_CONTEXT_KEY) + if fork_restored: + runtime.state["sandbox"] = Overwrite(runtime.state["sandbox"]) + original_state = runtime.state["sandbox"] + try: + with pytest.raises(SandboxRuntimeError, match="Thread ID not available"): + if async_path: + await ensure_sandbox_initialized_async(runtime) + else: + ensure_sandbox_initialized(runtime) + finally: + reset_sandbox_provider() + + assert provider.get_calls == [] + assert provider.scoped_get_calls == [] + assert provider.acquire_calls == [] + assert runtime.state["sandbox"] is original_state + assert "sandbox_id" not in runtime.context + + +def test_matching_checkpoint_sandbox_is_reused_without_acquire(): + provider = _IdentityScopedProvider() + set_sandbox_provider(provider) + runtime = _runtime_with_foreign_checkpoint() + runtime.state["sandbox"] = {"sandbox_id": OWN_SANDBOX_ID} + try: + sandbox = ensure_sandbox_initialized(runtime) + finally: + reset_sandbox_provider() + + assert sandbox.id == OWN_SANDBOX_ID + assert provider.acquire_calls == [] + assert provider.scoped_get_calls == [(OWN_SANDBOX_ID, "thread-a", "user-a")] + + +def test_aio_cached_lookup_requires_matching_user_and_thread_identity(): + provider = object.__new__(AioSandboxProvider) + provider._lock = threading.Lock() + foreign = _ScopedSandbox(FOREIGN_SANDBOX_ID) + provider._sandboxes = {FOREIGN_SANDBOX_ID: foreign} + provider._thread_sandboxes = {("user-b", "thread-b"): FOREIGN_SANDBOX_ID} + provider._active_sandbox_identity = {FOREIGN_SANDBOX_ID: ("user-b", "thread-b")} + provider._last_activity = {} + + assert ( + provider.get_scoped( + FOREIGN_SANDBOX_ID, + thread_id="thread-a", + user_id="user-a", + ) + is None + ) + assert ( + provider.get_scoped( + FOREIGN_SANDBOX_ID, + thread_id="thread-b", + user_id="user-b", + ) + is foreign + ) + assert FOREIGN_SANDBOX_ID in provider._last_activity