feat(authz): enforce sandbox:execute authorization at sandbox acquisition (#4063 Phase 3) (#4911)

Sandbox is an execution environment, not a named resource: multiple tools
(bash, read_file, write_file, glob, grep, ...) depend on it, all funneled
through ensure_sandbox_initialized / ensure_sandbox_initialized_async. Gate
the single acquisition entry point (single source of truth) instead of
maintaining a sandbox-tool-name set in middleware:

- authorize_sandbox_execution helper (authz/sandbox_authz.py) checks
  authorize("sandbox", "execute", target="*") — a binary judgment
  (can this role use the sandbox at all); RBAC allow:"*"/true permits,
  allow:[]/false denies.
- lazy path: ensure_sandbox_initialized (+ async) calls the gate before
  provider.acquire.
- eager path: SandboxMiddleware.before_agent / abefore_agent call the gate
  before _acquire_sandbox.
- deny raises SandboxAuthorizationError (SandboxError subclass) which
  propagates through tool execution as a friendly ToolMessage (RFC §9:
  'not a crash').
- authorization.enabled: false is a no-op everywhere; provider errors
  follow fail_closed (deny) / fail_open (allow).

12 tests in tests/test_sandbox_authorization.py cover disabled/allow/deny/
deny-via-bool/no-policy-unrestricted/provider-error-fail-closed/open/
internal-caller + ensure_sandbox_initialized deny (never acquires) and
allow (acquires) integration paths.

README.md

@@ -901,7 +901,7 @@ uv run python -m deerflow.skills.review.cli ../skills/public/data-analysis --for
 
 Tools follow the same philosophy. DeerFlow comes with a core toolset — web search, web fetch, rendered web capture, file operations, bash execution — and supports custom tools via MCP servers and Python functions. Swap anything. Add anything.
 
-Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. The built-in RBAC provider supports per-role `tools` and `routes` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md).
+Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md).
 
 Advanced deployments can also extend the agent runtime itself by declaring zero-argument `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. DeerFlow loads the same configured class list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, and broken modules fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Per-context parameterization and separate lead-only/subagent-only middleware lists are not supported yet.
 

backend/app/gateway/authz.py

@@ -46,6 +46,7 @@ from deerflow.config.authorization_config import AuthorizationConfig
 
 if TYPE_CHECKING:
     from app.gateway.auth.models import User
+    from deerflow.config.app_config import AppConfig
 
 P = ParamSpec("P")
 T = TypeVar("T")
@@ -299,23 +300,121 @@ def resolve_model_authorization(user: User, *, is_internal: bool) -> tuple[Autho
         logger.warning("Failed to resolve authorization provider for model routes", exc_info=True)
         raise _AuthorizationUnavailable(fail_closed=config.fail_closed)
 
+    principal = build_principal_from_context(
+        _route_authz_context(user, is_internal=is_internal),
+        default_role=config.default_role,
+    )
+    return provider, principal
+
+
+def _route_authz_context(user: User, *, is_internal: bool) -> dict:
+    """Build the shared Principal context dict for a request-scoped user.
+
+    Applies the ``INTERNAL_SYSTEM_ROLE → None`` pop so internal callers fall
+    under ``default_role`` (mirrors ``inject_authenticated_user_context``).
+    Used by ``resolve_model_authorization`` and ``authorize_sandbox_for_request``
+    so every route-level authorization path builds the identity the same way.
+    """
     from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE
 
     user_role = getattr(user, "system_role", None)
     if user_role == INTERNAL_SYSTEM_ROLE:
         user_role = None
+    return {
+        "user_id": str(user.id),
+        "user_role": user_role,
+        "oauth_provider": getattr(user, "oauth_provider", None),
+        "oauth_id": getattr(user, "oauth_id", None),
+        "is_internal": is_internal,
+    }
 
-    principal = build_principal_from_context(
-        {
-            "user_id": str(user.id),
-            "user_role": user_role,
-            "oauth_provider": getattr(user, "oauth_provider", None),
-            "oauth_id": getattr(user, "oauth_id", None),
-            "is_internal": is_internal,
-        },
-        default_role=config.default_role,
-    )
-    return provider, principal
+
+def authorize_sandbox_for_request(
+    user: User,
+    *,
+    is_internal: bool,
+    app_config: AppConfig | None,
+) -> None:
+    """Check ``sandbox:execute`` for a Gateway request before sandbox acquisition.
+
+    Thin wrapper over the harness-level ``authorize_sandbox_execution`` that
+    builds the Principal from the request-scoped ``user`` — the same identity
+    construction as ``resolve_model_authorization`` (including the
+    ``INTERNAL_SYSTEM_ROLE → None`` pop). Raises
+    :class:`~deerflow.sandbox.exceptions.SandboxAuthorizationError` on deny or
+    on provider-resolution failure under ``fail_closed``; callers translate
+    that into skipping the sandbox sync (not an HTTP error, since the primary
+    operation — e.g. file upload — can proceed without it).
+
+    No-op when ``authorization.enabled`` is false.
+    """
+    from deerflow.authz.sandbox_authz import authorize_sandbox_execution
+    from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+    config = _get_route_authorization_config()
+    if config.enabled is not True:
+        return
+
+    context = _route_authz_context(user, is_internal=is_internal)
+
+    try:
+        authorize_sandbox_execution(
+            context=context,
+            app_config=app_config,
+        )
+    except SandboxAuthorizationError:
+        raise
+    except Exception:
+        # Defense-in-depth: provider resolution and authorize() errors are
+        # already converted to SandboxAuthorizationError (or allowed under
+        # fail-open) one layer down inside authorize_sandbox_execution, so this
+        # normally only catches config-read failures here (e.g. get_config()
+        # raising in a config-less environment). Those must not 500 the
+        # upload/artifact route — degrade per fail_closed instead.
+        logger.warning("Failed to resolve authorization provider for sandbox:execute", exc_info=True)
+        if config.fail_closed:
+            raise SandboxAuthorizationError(role=context.get("user_role")) from None
+
+
+async def try_acquire_sandbox_for_request(
+    request: Request,
+    sandbox_provider,
+    thread_id: str,
+    *,
+    user_id: str,
+    app_config: AppConfig | None,
+) -> tuple[object, str | None, bool]:
+    """Gate + acquire the thread sandbox for a Gateway sync path.
+
+    Single entry point for the uploads/artifacts sandbox-sync paths so the
+    deny/skip semantics live in one place: runs the ``sandbox:execute`` gate
+    for the request's user, then acquires the sandbox. Returns
+    ``(sandbox, sandbox_id, denied)``:
+
+    - denied role → ``(None, None, True)``: acquisition was skipped by policy;
+      the primary operation (upload / artifact edit) proceeds without the
+      sandbox copy.
+    - allowed → ``(sandbox, sandbox_id, False)``: ``sandbox`` is the acquired
+      instance (``sandbox_id`` for later release), or ``sandbox is None`` when
+      the provider lost it right after acquiring (infrastructure error —
+      callers surface it as 500 / RuntimeError respectively, since that is
+      not a policy decision).
+    - ``request is None`` (direct-call tests) and unresolvable users skip the
+      gate — same fail-open semantics as the models routes' anonymous bypass.
+    """
+    from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+    try:
+        from app.gateway.deps import get_optional_user_from_request
+
+        user = await get_optional_user_from_request(request) if request is not None else None
+        if user is not None:
+            authorize_sandbox_for_request(user, is_internal=_is_internal_caller(request, user), app_config=app_config)
+    except SandboxAuthorizationError:
+        logger.info("Sandbox sync skipped: sandbox execution not permitted for this caller (thread_id=%s)", thread_id)
+        return None, None, True
+    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=user_id)
+    return sandbox_provider.get(sandbox_id), sandbox_id, False
 
 
 async def _authenticate(request: Request) -> AuthContext:

backend/app/gateway/routers/artifacts.py

@@ -16,10 +16,11 @@ from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import FileResponse, Response
 from pydantic import BaseModel, Field
 
-from app.gateway.authz import require_permission
+from app.gateway.authz import require_permission, try_acquire_sandbox_for_request
 from app.gateway.deps import get_run_manager
 from app.gateway.internal_auth import get_trusted_internal_owner_user_id
 from app.gateway.path_utils import resolve_thread_virtual_path
+from deerflow.authz.sandbox_authz import safe_app_config
 from deerflow.config.paths import make_safe_user_id
 from deerflow.runtime import ConflictError, ThreadOperationKind
 from deerflow.runtime.user_context import get_effective_user_id
@@ -498,7 +499,14 @@ async def update_artifact(
     body: ArtifactUpdateRequest,
     request: Request,
 ) -> ArtifactUpdateResponse:
-    """Update an existing text artifact while the thread has no active run."""
+    """Update an existing text artifact while the thread has no active run.
+
+    The host-side artifact file is updated first; when the sandbox provider is
+    not thread-mounted, the new content is also synced into the thread's
+    sandbox. Under ``authorization.enabled``, a caller denied
+    ``sandbox:execute`` skips that sandbox sync (the host-side update still
+    completes).
+    """
     virtual_path = _normalize_editable_artifact_path(path)
     raw_owner_user_id = get_trusted_internal_owner_user_id(request)
     effective_user_id = make_safe_user_id(raw_owner_user_id) if raw_owner_user_id else get_effective_user_id()
@@ -524,9 +532,18 @@ async def update_artifact(
 
             sandbox_provider = get_sandbox_provider()
             if not bool(getattr(sandbox_provider, "uses_thread_data_mounts", False)):
-                sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-                sandbox = sandbox_provider.get(sandbox_id)
-                if sandbox is None:
+                # Phase 3: enforce sandbox:execute before acquiring — a denied
+                # role skips the sandbox sync; the host-side artifact update
+                # still completes (the agent cannot consume the sandbox copy
+                # anyway when sandbox execution is denied).
+                sandbox, sandbox_id, sandbox_denied = await try_acquire_sandbox_for_request(
+                    request,
+                    sandbox_provider,
+                    thread_id,
+                    user_id=effective_user_id,
+                    app_config=safe_app_config(),
+                )
+                if not sandbox_denied and sandbox is None:
                     raise RuntimeError("Failed to acquire sandbox for artifact update")
 
             try:

backend/app/gateway/routers/uploads.py

@@ -11,7 +11,7 @@ from typing import BinaryIO
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 from pydantic import BaseModel, Field
 
-from app.gateway.authz import require_permission
+from app.gateway.authz import require_permission, try_acquire_sandbox_for_request
 from app.gateway.deps import get_config
 from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
@@ -305,7 +305,14 @@ async def upload_files(
     files: list[UploadFile] = File(...),
     config: AppConfig = Depends(get_config),
 ) -> UploadResponse:
-    """Upload multiple files to a thread's uploads directory."""
+    """Upload multiple files to a thread's uploads directory.
+
+    When the sandbox provider is not thread-mounted, uploaded files are also
+    synced into the thread's sandbox. Under ``authorization.enabled``, a caller
+    denied ``sandbox:execute`` skips that sync (the upload itself still
+    succeeds — files stay in the uploads dir; a sandbox-denied agent cannot
+    consume them anyway).
+    """
     if not files:
         raise HTTPException(status_code=400, detail="No files provided")
 
@@ -333,9 +340,19 @@ async def upload_files(
     sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider)
     sandbox = None
     if sync_to_sandbox:
-        sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-        sandbox = sandbox_provider.get(sandbox_id)
-        if sandbox is None:
+        # Phase 3: enforce sandbox:execute before acquiring — a role denied
+        # sandbox execution must not trigger sandbox allocation just by
+        # uploading files. Deny skips the sync; the upload itself still
+        # succeeds (files stay in the thread uploads dir; the agent cannot
+        # consume them via sandbox anyway).
+        sandbox, _sandbox_id, sandbox_denied = await try_acquire_sandbox_for_request(
+            request,
+            sandbox_provider,
+            thread_id,
+            user_id=effective_user_id,
+            app_config=config,
+        )
+        if not sandbox_denied and sandbox is None:
             raise HTTPException(status_code=500, detail="Failed to acquire sandbox")
     auto_convert_documents = _auto_convert_documents_enabled(config)
 
@@ -427,7 +444,7 @@ async def upload_files(
     # configuration can read the uploaded content.
     await run_file_io(_make_uploaded_paths_sandbox_readable, written_paths)
 
-    if sync_to_sandbox:
+    if sync_to_sandbox and sandbox is not None:
         for file_path, virtual_path in sandbox_sync_targets:
             await run_file_io(_sync_upload_to_sandbox, sandbox, file_path, virtual_path)
 

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -67,6 +67,10 @@ Authorization identity plumbing is independent of whether authorization enforcem
 
 Gateway route authorization uses `authz.py::resolve_route_permissions()` as the single provider integration point for both `AuthMiddleware` and decorator-only authentication. When enabled, it evaluates the six registered `threads:*` / `runs:*` permissions as `resource="route"` requests whose targets are the full `resource:action` strings. Decisions use the async provider API and are cached for the request in `AuthContext`; decorators do not call the provider again. Provider resolution or decision errors follow `authorization.fail_closed`, scoped per permission for decision errors. When authorization is disabled, the legacy complete permission set is returned without resolving a provider. Existing `owner_check` enforcement and `require_admin_user()` management gates remain independent and unchanged. Tests: `tests/test_authorization_route_permissions.py`, `tests/test_auth.py`, and `tests/test_auth_middleware.py`.
 
+Model authorization uses `authz.py::resolve_model_authorization()` (same cached-provider, internal-role, and principal-building path as route authorization) as the Gateway integration point for the `models` router: `list_models` filters names through `filter_resources(principal, "model", ...)`, and `get_model` enforces `authorize(resource="model", action="use")` with a deny surfacing as 403; provider errors follow `authorization.fail_closed` (fail-open returns the unfiltered list / proceeds). At runtime, `lead_agent/agent.py::_authorize_model_name` — called from `_make_lead_agent` and from `DeerFlowClient._ensure_agent` — applies the same `model:use` check to the resolved model name. On deny it scans the `filter_resources`-visible names (excluding the denied model), re-verifying each candidate with `authorize("model", "use")` before falling back, because a custom provider may allow `list` while denying `use`; no usable fallback raises under `fail_closed` and keeps the original model under fail-open. The built-in RBAC provider maps this to the per-role `models` policy key. Tests: `tests/test_models_authorization.py`.
+
+Sandbox authorization (`sandbox:execute`) gates every sandbox acquisition before `provider.acquire` — including this middleware's eager path (`before_agent` / `abefore_agent` skip acquisition on deny instead of raising, deferring to the lazy per-tool gate). See the [sandbox module guide](../../sandbox/AGENTS.md) authorization-gate paragraph and `tests/test_sandbox_authorization.py`.
+
 Before changing a later authorization phase, read the [authorization RFC](../../../../../../docs/plans/2026-07-10-pluggable-authorization-rfc.md) and its [implementation notes](../../../../../../docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md). The notes are the cumulative handoff record for merged PR behavior, reviewer feedback, trust-boundary decisions, deferred scope, and required regression coverage.
 
 **Lead-only middlewares** (`build_middlewares`, appended after the base):

backend/packages/harness/deerflow/authz/__init__.py

@@ -6,6 +6,7 @@ from deerflow.authz.principal import build_principal_from_context, normalize_aut
 from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzReason, AuthzRequest, Principal
 from deerflow.authz.rbac import RbacAuthorizationProvider
 from deerflow.authz.runtime import resolve_authorization_provider
+from deerflow.authz.sandbox_authz import authorize_sandbox_execution
 from deerflow.authz.tool_filter import apply_tool_authorization
 
 __all__ = [
@@ -17,6 +18,7 @@ __all__ = [
     "Principal",
     "RbacAuthorizationProvider",
     "apply_tool_authorization",
+    "authorize_sandbox_execution",
     "build_principal_from_context",
     "filter_tools_by_authorization",
     "normalize_authz_attributes",

backend/packages/harness/deerflow/authz/sandbox_authz.py

@@ -0,0 +1,101 @@
+"""Sandbox execution authorization gate.
+
+Checks ``authorize("sandbox", "execute")`` before sandbox acquisition so a
+role-scoped policy can deny sandbox execution entirely. On deny, a
+:class:`~deerflow.sandbox.exceptions.SandboxAuthorizationError` propagates up
+through the tool's execution; the agent's tool-error handling converts it to a
+friendly ``ToolMessage`` ("sandbox not permitted for your role") rather than
+crashing the run (RFC §9).
+
+Mirrors the Principal/provider pattern of ``apply_tool_authorization``
+(``tool_filter.py``) and ``_authorize_model_name`` (``lead_agent/agent.py``)
+so the sandbox path shares one identity source with the tool and model paths.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+from typing import Any
+
+from deerflow.authz.principal import build_principal_from_context
+from deerflow.authz.provider import AuthzDecision, AuthzRequest
+from deerflow.authz.runtime import resolve_authorization_provider
+from deerflow.config.app_config import AppConfig
+from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+logger = logging.getLogger(__name__)
+
+
+def safe_app_config() -> AppConfig | None:
+    """Load the global AppConfig, returning None when unavailable.
+
+    Authorization can only be enabled via config, so no readable config ⇒ the
+    sandbox gate is a no-op (``authorize_sandbox_execution`` treats a ``None``
+    app_config the same as ``authorization.enabled: false``). This keeps the
+    gate safe in environments without a ``config.yaml`` (e.g. CI runners and
+    direct-call tests) where ``get_app_config`` would raise ``FileNotFoundError``.
+    """
+    try:
+        from deerflow.config import get_app_config
+
+        return get_app_config()
+    except Exception:
+        logger.debug("App config unavailable; sandbox:execute gate is a no-op", exc_info=True)
+        return None
+
+
+# Sandbox is a single shared resource (the execution environment), not a named
+# catalog like tools/models/skills. The target is therefore a sentinel "*" that
+# means "the sandbox as a whole"; RBAC ``allow: ["*"]`` / ``allow: true`` permits
+# it, ``allow: []`` / ``allow: false`` denies it.
+_SANDBOX_TARGET = "*"
+
+
+def authorize_sandbox_execution(*, context: Mapping[str, Any], app_config: AppConfig | None) -> None:
+    """Check ``authorize("sandbox", "execute")`` before sandbox acquisition.
+
+    ``app_config=None`` (unreadable config) is treated the same as
+    ``authorization.enabled: false`` — a no-op. On deny (or provider error
+    with ``fail_closed``), raises :class:`SandboxAuthorizationError`; on
+    provider error with fail-open, returns silently (legacy allow behavior).
+    """
+    # Guard against Mock/SimpleNamespace app_config objects in tests that
+    # don't carry a real AuthorizationConfig. getattr avoids AttributeError
+    # and the ``is not True`` identity check avoids truthy Mock attributes
+    # (mirrors filter_available_skills_by_authorization in skill_filter.py).
+    authz_config = getattr(app_config, "authorization", None)
+    if authz_config is None or getattr(authz_config, "enabled", None) is not True:
+        return
+
+    # Provider *resolution* failures follow the same fail_closed/fail_open
+    # decision as authorize() errors — a raw ValueError here would otherwise
+    # effectively deny under fail_open (inverted semantics).
+    try:
+        provider = resolve_authorization_provider(authz_config)
+    except Exception:
+        logger.warning("Failed to resolve authorization provider for sandbox:execute", exc_info=True)
+        if authz_config.fail_closed:
+            raise SandboxAuthorizationError() from None
+        # fail-open: allow sandbox acquisition despite the resolution error.
+        return
+    if provider is None:
+        return
+
+    principal = build_principal_from_context(context, default_role=authz_config.default_role)
+    try:
+        decision = provider.authorize(AuthzRequest(principal=principal, resource="sandbox", action="execute", target=_SANDBOX_TARGET))
+        if not isinstance(decision, AuthzDecision):
+            raise TypeError("AuthorizationProvider.authorize must return AuthzDecision")
+        if decision.allow:
+            return
+        # Explicit deny → block sandbox acquisition with a friendly error.
+        raise SandboxAuthorizationError(role=principal.role)
+    except SandboxAuthorizationError:
+        raise
+    except Exception:
+        logger.warning("Authorization provider failed while checking sandbox:execute", exc_info=True)
+        if authz_config.fail_closed:
+            raise SandboxAuthorizationError(role=principal.role)
+        # fail-open: allow sandbox acquisition despite the provider error.
+        return

backend/packages/harness/deerflow/sandbox/AGENTS.md

@@ -2,6 +2,7 @@
 
 **Interface**: Abstract `Sandbox` with `execute_command(command, env=None)`, `read_file`, `write_file`, `list_dir`, `glob`, and `grep`. `grep` accepts either one text file or a directory tree. The optional `env` injects per-call environment variables (request-scoped secrets — see Request-Scoped Secrets below); `LocalSandbox` merges it via `subprocess.run(env=...)` and `AioSandbox` routes env-bearing commands through the `bash.exec(env=...)` API on a fresh session.
 **Provider Pattern**: `SandboxProvider` with `acquire`, `acquire_async`, `get`, `release` lifecycle. Async agent/tool paths call async sandbox lifecycle hooks so Docker sandbox creation, discovery, cross-process locking, readiness polling, and release stay off the event loop.
+**Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox acquisition passes through `authorize_sandbox_execution` (`deerflow/authz/sandbox_authz.py`) - a binary `authorize(principal, "sandbox", "execute", target="*")` check before `provider.acquire`. The gate lives at the single acquisition entry point (`ensure_sandbox_initialized` / `_acquire_sandbox_async` in `tools.py`, and `SandboxMiddleware.before_agent` / `abefore_agent`), so it cannot be bypassed regardless of which sandbox-dependent tool triggers it; the reuse path (sandbox already in state) skips the re-check. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of 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 (both `authorize()` 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 auxiliary sync paths (uploads/artifacts routers) call `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates via `authorize_sandbox_for_request` and skips the sync on deny - the upload/artifact edit itself still succeeds. Tests: `tests/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**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Public, custom, legacy, and managed integration skill mappings point at stable enabled-only projection roots rather than raw skill directories.

backend/packages/harness/deerflow/sandbox/exceptions.py

@@ -111,3 +111,24 @@ class SandboxCapacityExceededError(SandboxError):
         self.replicas = replicas
         self.retry_after_seconds = retry_after_seconds
         self.reason = reason
+
+
+class SandboxAuthorizationError(SandboxError):
+    """Raised when the caller's role is denied sandbox execution.
+
+    Phase 3 pluggable authorization: ``authorize("sandbox", "execute")`` is
+    checked before sandbox acquisition. On deny this error propagates up
+    through the tool's execution so the agent's tool-error handling converts
+    it to a friendly ``ToolMessage`` ("sandbox not permitted for your role"),
+    rather than crashing the run (RFC §9).
+    """
+
+    def __init__(
+        self,
+        message: str = "Sandbox execution is not permitted for your role",
+        *,
+        role: str | None = None,
+    ) -> None:
+        details = {"role": role} if role else None
+        super().__init__(message, details)
+        self.role = role

backend/packages/harness/deerflow/sandbox/middleware.py

@@ -12,8 +12,10 @@ from langgraph.runtime import Runtime
 from langgraph.types import Command
 
 from deerflow.agents.thread_state import SandboxStateField, ThreadDataState
+from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
 from deerflow.runtime.user_context import resolve_runtime_user_id
 from deerflow.sandbox import get_sandbox_provider
+from deerflow.sandbox.exceptions import SandboxAuthorizationError
 from deerflow.sandbox.overwrite import unwrap_sandbox
 
 logger = logging.getLogger(__name__)
@@ -76,6 +78,21 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
             thread_id = (runtime.context or {}).get("thread_id")
             if thread_id is None:
                 return super().before_agent(state, runtime)
+            # Phase 3: enforce sandbox:execute authorization before acquiring
+            # (eager path). On deny, skip the eager acquisition instead of
+            # raising: an exception here is outside any tool call, so it would
+            # surface as a run-level graph error rather than the RFC §9
+            # friendly ToolMessage. Skipping defers to the lazy gate inside
+            # ``ensure_sandbox_initialized``, which denies per-tool with the
+            # friendly message on the first sandbox-touching tool call.
+            try:
+                authorize_sandbox_execution(
+                    context=runtime.context or {},
+                    app_config=safe_app_config(),
+                )
+            except SandboxAuthorizationError:
+                logger.info("Sandbox execution denied for this role; skipping eager sandbox acquisition (thread_id=%s)", thread_id)
+                return None
             sandbox_id = self._acquire_sandbox(thread_id, user_id=resolve_runtime_user_id(runtime))
             logger.info(f"Assigned sandbox {sandbox_id} to thread {thread_id}")
             return {"sandbox": {"sandbox_id": sandbox_id}}
@@ -93,6 +110,19 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
             thread_id = (runtime.context or {}).get("thread_id")
             if thread_id is None:
                 return await super().abefore_agent(state, runtime)
+            # Phase 3: enforce sandbox:execute authorization before acquiring
+            # (eager path, async counterpart of the gate in before_agent). On
+            # deny, skip the eager acquisition — the lazy gate inside
+            # ``ensure_sandbox_initialized`` denies per-tool with the RFC §9
+            # friendly message on the first sandbox-touching tool call.
+            try:
+                authorize_sandbox_execution(
+                    context=runtime.context or {},
+                    app_config=safe_app_config(),
+                )
+            except SandboxAuthorizationError:
+                logger.info("Sandbox execution denied for this role; skipping eager sandbox acquisition (thread_id=%s)", thread_id)
+                return None
             sandbox_id = await self._acquire_sandbox_async(thread_id, user_id=resolve_runtime_user_id(runtime))
             logger.info(f"Assigned sandbox {sandbox_id} to thread {thread_id}")
             return {"sandbox": {"sandbox_id": sandbox_id}}

backend/packages/harness/deerflow/sandbox/tools.py

@@ -12,6 +12,7 @@ from pathlib import Path
 from langchain.tools import tool
 
 from deerflow.agents.thread_state import ThreadDataState
+from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
 from deerflow.config import get_app_config
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
 from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
@@ -1416,6 +1417,15 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
     if thread_id is None:
         raise SandboxRuntimeError("Thread ID not available in runtime context")
 
+    # Phase 3: enforce sandbox:execute authorization before acquiring. On deny
+    # a SandboxAuthorizationError propagates up through the tool so the agent's
+    # tool-error handling returns a friendly message (RFC §9). Skipped on the
+    # reuse path above (already authorized when first acquired).
+    authorize_sandbox_execution(
+        context=runtime.context or {},
+        app_config=safe_app_config(),
+    )
+
     provider = get_sandbox_provider()
     sandbox_id = provider.acquire(thread_id, user_id=resolve_runtime_user_id(runtime))
 
@@ -1463,6 +1473,13 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
     if thread_id is None:
         raise SandboxRuntimeError("Thread ID not available in runtime context")
 
+    # Phase 3: enforce sandbox:execute authorization before acquiring (async
+    # counterpart of the sync gate in ``ensure_sandbox_initialized``).
+    authorize_sandbox_execution(
+        context=runtime.context or {},
+        app_config=safe_app_config(),
+    )
+
     provider = get_sandbox_provider()
     sandbox_id = await provider.acquire_async(thread_id, user_id=resolve_runtime_user_id(runtime))
 

backend/tests/test_sandbox_authorization.py

@@ -0,0 +1,696 @@
+"""Phase 3 sandbox-level authorization tests.
+
+Covers the ``authorize("sandbox", "execute")`` gate at the sandbox-acquisition
+entry point. When denied, a :class:`SandboxAuthorizationError` propagates up
+through the tool so the agent's tool-error handling returns a friendly message
+(RFC §9), rather than crashing the run.
+
+The gate lives in :func:`authorize_sandbox_execution` (``authz/sandbox_authz.py``)
+and is called from:
+- ``ensure_sandbox_initialized`` / ``ensure_sandbox_initialized_async`` (lazy path)
+- ``SandboxMiddleware.before_agent`` / ``abefore_agent`` (eager path)
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from deerflow.authz.provider import AuthzDecision, AuthzReason
+from deerflow.authz.rbac import RbacAuthorizationProvider
+from deerflow.authz.sandbox_authz import authorize_sandbox_execution
+from deerflow.config.app_config import AppConfig
+from deerflow.config.authorization_config import AuthorizationConfig
+from deerflow.config.model_config import ModelConfig
+from deerflow.config.sandbox_config import SandboxConfig
+from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+# ── Helpers ────────────────────────────────────────────────────────────
+
+
+def _make_app_config() -> AppConfig:
+    """Build a minimal AppConfig for authorization tests."""
+    return AppConfig(
+        models=[ModelConfig(name="gpt-4", model="gpt-4", use="langchain_openai:ChatOpenAI")],
+        sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
+        authorization=AuthorizationConfig(),
+    )
+
+
+def _context(**overrides):
+    values = {
+        "user_id": "user-123",
+        "user_role": "user",
+        "oauth_provider": "github",
+        "oauth_id": "oauth-456",
+        "is_internal": False,
+    }
+    values.update(overrides)
+    return values
+
+
+def _enable_authz(app_config: AppConfig, *, fail_closed: bool = True, default_role: str = "user") -> None:
+    app_config.authorization = AuthorizationConfig(
+        enabled=True,
+        fail_closed=fail_closed,
+        default_role=default_role,
+    )
+
+
+# ── authorize_sandbox_execution unit tests ─────────────────────────────
+
+
+def test_authorize_sandbox_disabled_is_noop():
+    """When authorization is disabled, no check is performed (allow)."""
+    app_config = _make_app_config()
+    # AuthorizationConfig() defaults to enabled=False.
+    authorize_sandbox_execution(context=_context(), app_config=app_config)  # must not raise
+
+
+def test_authorize_sandbox_rbac_allow(monkeypatch):
+    """Role with sandbox allow → permitted."""
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": "*"}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    authorize_sandbox_execution(context=_context(), app_config=app_config)  # must not raise
+
+
+def test_authorize_sandbox_rbac_deny(monkeypatch):
+    """Role with sandbox allow: [] → SandboxAuthorizationError."""
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    with pytest.raises(SandboxAuthorizationError, match="not permitted for your role"):
+        authorize_sandbox_execution(context=_context(), app_config=app_config)
+
+
+def test_authorize_sandbox_rbac_deny_via_bool(monkeypatch):
+    """Role with sandbox allow: false → denied."""
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": False}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    with pytest.raises(SandboxAuthorizationError):
+        authorize_sandbox_execution(context=_context(), app_config=app_config)
+
+
+def test_authorize_sandbox_no_policy_is_unrestricted(monkeypatch):
+    """Role with no sandbox policy → unrestricted (allow)."""
+    provider = RbacAuthorizationProvider(roles={"user": {"tools": {"allow": "*"}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    authorize_sandbox_execution(context=_context(), app_config=app_config)  # must not raise
+
+
+def test_authorize_sandbox_provider_error_fail_closed(monkeypatch):
+    """Provider error + fail_closed → SandboxAuthorizationError."""
+
+    class _ErrorProvider:
+        name = "error"
+
+        def authorize(self, request):
+            raise RuntimeError("boom")
+
+        async def aauthorize(self, request):
+            raise RuntimeError("boom")
+
+        def filter_resources(self, principal, resource_type, candidates):
+            raise RuntimeError("boom")
+
+    app_config = _make_app_config()
+    _enable_authz(app_config, fail_closed=True)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: _ErrorProvider(),
+    )
+    with pytest.raises(SandboxAuthorizationError):
+        authorize_sandbox_execution(context=_context(), app_config=app_config)
+
+
+def test_authorize_sandbox_provider_error_fail_open(monkeypatch):
+    """Provider error + fail_open → allow (no raise)."""
+
+    class _ErrorProvider:
+        name = "error"
+
+        def authorize(self, request):
+            raise RuntimeError("boom")
+
+        async def aauthorize(self, request):
+            raise RuntimeError("boom")
+
+        def filter_resources(self, principal, resource_type, candidates):
+            raise RuntimeError("boom")
+
+    app_config = _make_app_config()
+    _enable_authz(app_config, fail_closed=False)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: _ErrorProvider(),
+    )
+    authorize_sandbox_execution(context=_context(), app_config=app_config)  # must not raise
+
+
+def test_authorize_sandbox_internal_caller_uses_default_role(monkeypatch):
+    """Internal callers (system_role=None) fall under default_role."""
+    provider = RbacAuthorizationProvider(
+        roles={
+            "user": {"sandbox": {"allow": []}},
+            "admin": {"sandbox": {"allow": "*"}},
+        }
+    )
+    app_config = _make_app_config()
+    _enable_authz(app_config, default_role="admin")
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    # Internal caller with system_role=None → default_role="admin" → allowed.
+    authorize_sandbox_execution(
+        context=_context(user_role=None, is_internal=True),
+        app_config=app_config,
+    )
+
+
+def test_authorize_sandbox_denied_error_carries_role(monkeypatch):
+    """The SandboxAuthorizationError carries the denied role for diagnostics."""
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    with pytest.raises(SandboxAuthorizationError) as exc_info:
+        authorize_sandbox_execution(context=_context(), app_config=app_config)
+    assert exc_info.value.role == "user"
+
+
+# ── Integration: ensure_sandbox_initialized denies on authz reject ──────
+
+
+def test_ensure_sandbox_initialized_denies_on_authz_reject(monkeypatch):
+    """ensure_sandbox_initialized raises SandboxAuthorizationError on deny.
+
+    Real-path test: the genuine gate runs inside ensure_sandbox_initialized
+    before provider.acquire is touched, so a denied role never acquires a
+    sandbox (and the error propagates as a friendly ToolMessage upstream).
+    """
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    runtime = SimpleNamespace(
+        state={"sandbox": None},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+    # A mock provider whose acquire must never be called.
+    sandbox_provider = MagicMock()
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+
+    with pytest.raises(SandboxAuthorizationError):
+        sandbox_tools.ensure_sandbox_initialized(runtime)
+    sandbox_provider.acquire.assert_not_called()
+
+
+def test_ensure_sandbox_initialized_allows_on_authz_permit(monkeypatch):
+    """ensure_sandbox_initialized proceeds to acquire on allow."""
+    from deerflow.sandbox import tools as sandbox_tools
+    from deerflow.sandbox.sandbox_provider import (
+        reset_sandbox_provider,
+        set_sandbox_provider,
+    )
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": "*"}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    acquired = {"called": False}
+
+    class _StubProvider:
+        def acquire(self, thread_id=None, *, user_id=None):
+            acquired["called"] = True
+            return "sbx-1"
+
+        def get(self, sandbox_id):
+            return MagicMock()  # sandbox instance shape doesn't matter for this test
+
+        def release(self, sandbox_id):
+            pass
+
+        def reset(self):
+            pass
+
+    set_sandbox_provider(_StubProvider())
+    try:
+        runtime = SimpleNamespace(
+            state={"sandbox": None},
+            context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+            config=None,
+        )
+        sandbox = sandbox_tools.ensure_sandbox_initialized(runtime)
+        assert sandbox is not None
+        assert acquired["called"] is True
+    finally:
+        reset_sandbox_provider()
+
+
+# ── Custom provider: authorize called with correct args ────────────────
+
+
+def test_authorize_sandbox_calls_provider_with_correct_request(monkeypatch):
+    """The provider receives resource='sandbox', action='execute', target='*'."""
+
+    class _RecordingProvider:
+        name = "recording"
+
+        def __init__(self):
+            self.request = None
+
+        def authorize(self, request):
+            self.request = request
+            return AuthzDecision(allow=True, reasons=[AuthzReason(code="authz.allowed")])
+
+        async def aauthorize(self, request):
+            return self.authorize(request)
+
+        def filter_resources(self, principal, resource_type, candidates):
+            return list(candidates)
+
+    provider = _RecordingProvider()
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    authorize_sandbox_execution(context=_context(), app_config=app_config)
+
+    assert provider.request.resource == "sandbox"
+    assert provider.request.action == "execute"
+    assert provider.request.target == "*"
+
+
+# ── Gateway auxiliary paths: uploads / artifacts sync ──────────────────
+# Regression for the multi-path coverage self-check (pr-review checkpoint 14):
+# the uploads and artifacts routers acquire the thread sandbox directly (not
+# via ensure_sandbox_initialized) to sync files. A role denied sandbox:execute
+# must not trigger sandbox allocation there; the primary operation (upload /
+# artifact edit) still succeeds with the sandbox sync skipped.
+
+
+def _make_upload_app(monkeypatch, provider, *, fail_closed: bool = True):
+    """Build a FastAPI app with the uploads router and authz enabled."""
+    from _router_auth_helpers import make_authed_test_app
+    from fastapi.testclient import TestClient
+
+    from app.gateway.routers import uploads as uploads_router
+
+    app_config = _make_app_config()
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=fail_closed, default_role="user")
+    monkeypatch.setattr(
+        "app.gateway.authz._get_route_authorization_config",
+        lambda: app_config.authorization,
+    )
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+
+    app = make_authed_test_app()
+    app.include_router(uploads_router.router)
+    app.dependency_overrides[uploads_router.get_config] = lambda: app_config
+    return TestClient(app)
+
+
+def test_upload_sandbox_sync_skipped_when_denied(monkeypatch, tmp_path):
+    """Denied role: upload succeeds, sandbox.acquire is never called."""
+    from unittest.mock import AsyncMock, MagicMock
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    client = _make_upload_app(monkeypatch, provider)
+    _isolated_uploads_dir(monkeypatch, tmp_path)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.uses_thread_data_mounts = False
+    sandbox_provider.acquire_async = AsyncMock(side_effect=AssertionError("must not acquire"))
+    monkeypatch.setattr("app.gateway.routers.uploads.get_sandbox_provider", lambda: sandbox_provider)
+    monkeypatch.setattr(
+        "app.gateway.deps.get_optional_user_from_request",
+        AsyncMock(return_value=_request_user()),
+    )
+
+    resp = client.post("/api/threads/upload-test/uploads", files={"files": ("a.txt", b"hello")})
+    assert resp.status_code == 200, resp.text
+    sandbox_provider.acquire_async.assert_not_called()
+
+
+def test_upload_sandbox_sync_proceeds_when_allowed(monkeypatch, tmp_path):
+    """Allowed role: sandbox.acquire_async is called as before."""
+    from unittest.mock import AsyncMock, MagicMock
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": "*"}}})
+    client = _make_upload_app(monkeypatch, provider)
+    _isolated_uploads_dir(monkeypatch, tmp_path)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.uses_thread_data_mounts = False
+    sandbox_provider.acquire_async = AsyncMock(return_value="sbx-1")
+    sandbox_provider.get = MagicMock(return_value=MagicMock())
+    monkeypatch.setattr("app.gateway.routers.uploads.get_sandbox_provider", lambda: sandbox_provider)
+    monkeypatch.setattr(
+        "app.gateway.deps.get_optional_user_from_request",
+        AsyncMock(return_value=_request_user()),
+    )
+
+    resp = client.post("/api/threads/upload-test/uploads", files={"files": ("a.txt", b"hello")})
+    assert resp.status_code == 200, resp.text
+    sandbox_provider.acquire_async.assert_called_once()
+
+
+def _isolated_uploads_dir(monkeypatch, tmp_path):
+    """Redirect thread uploads storage to tmp_path (test-isolation).
+
+    Without this the upload route writes into the real global uploads root,
+    polluting other tests that assert on uploads directory state.
+    """
+    uploads_dir = tmp_path / "uploads"
+    uploads_dir.mkdir(parents=True)
+    monkeypatch.setattr("app.gateway.routers.uploads.get_uploads_dir", lambda thread_id, user_id=None: uploads_dir)
+    monkeypatch.setattr("app.gateway.routers.uploads.ensure_uploads_dir", lambda thread_id, user_id=None: uploads_dir)
+    return uploads_dir
+
+
+def _request_user():
+    from types import SimpleNamespace
+
+    return SimpleNamespace(id="user-123", system_role="user", oauth_provider=None, oauth_id=None)
+
+
+def test_authorize_sandbox_mock_app_config_is_noop():
+    """A SimpleNamespace/duck-typed app_config without `authorization` is a no-op.
+
+    Mirrors the mock-safe guard in filter_available_skills_by_authorization
+    (skill_filter.py) — regression for the consistency finding willem-bd raised
+    on apply_skill_authorization in #4541: the sibling guards use getattr so
+    test doubles that don't carry a real AuthorizationConfig don't blow up.
+    """
+    authorize_sandbox_execution(context=_context(), app_config=SimpleNamespace(sandbox=None))  # must not raise
+
+
+# ── willem-bd review round 1 regressions ───────────────────────────────
+
+
+def test_authorize_sandbox_resolution_error_fail_open_allows(monkeypatch):
+    """Provider *resolution* error + fail_open → allow (not an inverted deny).
+
+    Regression for willem-bd's finding: resolve_authorization_provider ran
+    outside the try, so a misconfigured provider (ValueError) propagated as a
+    raw exception and effectively denied under fail_open — inverted semantics.
+    """
+
+    def _boom(_config):
+        raise ValueError("bad provider class path")
+
+    app_config = _make_app_config()
+    _enable_authz(app_config, fail_closed=False)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        _boom,
+    )
+    authorize_sandbox_execution(context=_context(), app_config=app_config)  # must not raise
+
+
+def test_authorize_sandbox_resolution_error_fail_closed_denies(monkeypatch):
+    """Provider *resolution* error + fail_closed → SandboxAuthorizationError."""
+
+    def _boom(_config):
+        raise ValueError("bad provider class path")
+
+    app_config = _make_app_config()
+    _enable_authz(app_config, fail_closed=True)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        _boom,
+    )
+    with pytest.raises(SandboxAuthorizationError):
+        authorize_sandbox_execution(context=_context(), app_config=app_config)
+
+
+def test_eager_before_agent_deny_skips_acquisition(monkeypatch):
+    """Eager-path deny skips acquisition instead of raising a run-level error.
+
+    Regression for willem-bd's finding: an exception from before_agent is
+    outside any tool call, so it would surface as a graph error rather than
+    the RFC §9 friendly ToolMessage. On deny the middleware now returns None
+    (no sandbox assigned); the lazy gate denies per-tool later.
+    """
+    from types import SimpleNamespace as _NS
+
+    from deerflow.sandbox.middleware import SandboxMiddleware
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    acquired = {"called": False}
+
+    def _no_acquire(_thread_id, *, user_id=None):
+        acquired["called"] = True
+        return "sbx-1"
+
+    middleware = SandboxMiddleware(lazy_init=False)
+    monkeypatch.setattr(middleware, "_acquire_sandbox", _no_acquire)
+
+    runtime = _NS(context={"thread_id": "t1", "user_id": "u1", "user_role": "user"})
+    result = middleware.before_agent({"sandbox": None}, runtime)
+
+    assert result is None  # no sandbox assigned
+    assert acquired["called"] is False  # acquire never touched
+
+
+def test_artifact_sandbox_sync_skipped_when_denied(monkeypatch, tmp_path):
+    """Artifacts router deny path: host-side update completes, acquire skipped.
+
+    Regression for willem-bd's round-2 nit — the uploads router's deny path
+    had coverage but this copy didn't; a reverted gate here would pass the
+    suite unnoticed.
+    """
+    from unittest.mock import AsyncMock, MagicMock
+
+    from app.gateway.routers import artifacts as artifacts_router
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    app_config.authorization = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+    monkeypatch.setattr(
+        "app.gateway.authz._get_route_authorization_config",
+        lambda: app_config.authorization,
+    )
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.uses_thread_data_mounts = False
+    sandbox_provider.acquire_async = AsyncMock(side_effect=AssertionError("must not acquire"))
+    monkeypatch.setattr(artifacts_router, "get_sandbox_provider", lambda: sandbox_provider)
+    monkeypatch.setattr(artifacts_router, "resolve_thread_virtual_path", lambda _t, _p, user_id=None: tmp_path / "note.txt")
+
+    from contextlib import asynccontextmanager
+
+    @asynccontextmanager
+    async def _allow_write(*_args, **_kwargs):
+        yield
+
+    monkeypatch.setattr(artifacts_router, "reserve_artifact_write", _allow_write)
+    monkeypatch.setattr(
+        "app.gateway.deps.get_optional_user_from_request",
+        AsyncMock(return_value=_request_user()),
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    from _router_auth_helpers import call_unwrapped
+
+    artifact_path = tmp_path / "note.txt"
+    artifact_path.write_bytes(b"before")
+    request = type("R", (), {})()  # simple object; only passed through to mocked helpers
+
+    import asyncio
+    import hashlib
+
+    from app.gateway.routers.artifacts import ArtifactUpdateRequest
+
+    sha = hashlib.sha256(b"before").hexdigest()
+    asyncio.run(
+        call_unwrapped(
+            artifacts_router.update_artifact,
+            "t-denied",
+            "mnt/user-data/outputs/note.txt",
+            ArtifactUpdateRequest(content="after", expected_sha256=sha),
+            request,
+        )
+    )
+    sandbox_provider.acquire_async.assert_not_called()
+    assert artifact_path.read_bytes() == b"after"
+
+
+def test_authorize_sandbox_no_config_file_is_noop(monkeypatch):
+    """No readable config.yaml (CI environments) → the gate is a no-op.
+
+    Regression for the CI failure: get_app_config raises FileNotFoundError in
+    config-less environments, which previously propagated out of the gate and
+    broke ensure_sandbox_initialized's direct-call tests.
+    """
+    from deerflow.authz import sandbox_authz as mod
+
+    def _no_config():
+        raise FileNotFoundError("config.yaml file not found in the project root")
+
+    monkeypatch.setattr("deerflow.config.get_app_config", _no_config)
+    # safe_app_config imports get_app_config lazily from deerflow.config.
+    assert mod.safe_app_config() is None
+    # And the gate itself tolerates app_config=None (same as disabled).
+    authorize_sandbox_execution(context=_context(), app_config=None)  # must not raise
+
+
+def test_upload_gate_tolerates_request_none(monkeypatch, tmp_path):
+    """Direct-call tests pass request=None; the uploads gate must not crash.
+
+    Regression for the blocking-io CI failure: get_optional_user_from_request
+    dereferences request.cookies on the no-state-user path, so request=None
+    raised AttributeError before the None guard.
+    """
+    import asyncio
+    from unittest.mock import AsyncMock, MagicMock
+
+    from app.gateway.routers import uploads as uploads_router
+
+    _isolated_uploads_dir(monkeypatch, tmp_path)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.uses_thread_data_mounts = False
+    sandbox_provider.acquire_async = AsyncMock(return_value="sbx-1")
+    sandbox_provider.get = MagicMock(return_value=MagicMock())
+    monkeypatch.setattr(uploads_router, "get_sandbox_provider", lambda: sandbox_provider)
+    # If the request=None guard were missing, this lookup would dereference
+    # request.cookies and raise AttributeError; with the guard it is never called.
+    monkeypatch.setattr(
+        "app.gateway.deps.get_optional_user_from_request",
+        AsyncMock(side_effect=AttributeError("'NoneType' object has no attribute 'cookies'")),
+    )
+
+    from io import BytesIO
+
+    from _router_auth_helpers import call_unwrapped
+    from fastapi import UploadFile
+
+    config = SimpleNamespace(uploads={"max_files": 5, "max_file_size": 10**6})
+    file = UploadFile(filename="a.txt", file=BytesIO(b"hello"))
+    # request=None — the None guard must skip the user lookup entirely.
+    result = asyncio.run(call_unwrapped(uploads_router.upload_files, "t-none", request=None, files=[file], config=config))
+    assert result.success is True
+    sandbox_provider.acquire_async.assert_called_once()
+
+
+def test_ensure_sandbox_initialized_async_denies_on_authz_reject(monkeypatch):
+    """Async acquisition path deny: acquire_async never called.
+
+    Regression for willem-bd's round-4 finding — the async gate copy is
+    verbatim-sync, so without this test deleting it would leave the suite green.
+    """
+    from deerflow.sandbox import tools as sandbox_tools
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    sandbox_provider = MagicMock()
+    sandbox_provider.acquire_async = AsyncMock(side_effect=AssertionError("must not acquire"))
+    monkeypatch.setattr(sandbox_tools, "get_sandbox_provider", lambda: sandbox_provider)
+
+    runtime = SimpleNamespace(
+        state={"sandbox": None},
+        context={"thread_id": "t1", "user_id": "u1", "user_role": "user"},
+        config=None,
+    )
+    import asyncio
+
+    with pytest.raises(SandboxAuthorizationError):
+        asyncio.run(sandbox_tools.ensure_sandbox_initialized_async(runtime))
+    sandbox_provider.acquire_async.assert_not_called()
+
+
+def test_abefore_agent_deny_skips_acquisition(monkeypatch):
+    """Async eager path deny: acquisition skipped, no run-level error.
+
+    Async counterpart of test_eager_before_agent_deny_skips_acquisition.
+    """
+    from deerflow.sandbox.middleware import SandboxMiddleware
+
+    provider = RbacAuthorizationProvider(roles={"user": {"sandbox": {"allow": []}}})
+    app_config = _make_app_config()
+    _enable_authz(app_config)
+    monkeypatch.setattr(
+        "deerflow.authz.sandbox_authz.resolve_authorization_provider",
+        lambda config: provider,
+    )
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: app_config)
+
+    acquired = {"called": False}
+
+    async def _no_acquire(_thread_id, *, user_id=None):
+        acquired["called"] = True
+        return "sbx-1"
+
+    middleware = SandboxMiddleware(lazy_init=False)
+    monkeypatch.setattr(middleware, "_acquire_sandbox_async", _no_acquire)
+
+    runtime = SimpleNamespace(context={"thread_id": "t1", "user_id": "u1", "user_role": "user"})
+    import asyncio
+
+    result = asyncio.run(middleware.abefore_agent({"sandbox": None}, runtime))
+
+    assert result is None  # no sandbox assigned
+    assert acquired["called"] is False  # acquire_async never touched

config.example.yaml

@@ -2484,14 +2484,21 @@ run_ownership:
 #           tools: {allow: "*"}
 #           routes: {allow: "*"}
 #           models: {allow: "*"}
+#           sandbox: {allow: "*"}
 #         user:
 #           tools: {allow: "*", deny: ["update_agent"]}
 #           routes: {allow: "*"}
 #           models: {allow: "*"}
+#           sandbox: {allow: "*"}
 #         guest:
-#           tools: {allow: ["web_search", "read_file"]}
+#           # web-only role: sandbox-dependent tools (read_file, bash, glob,
+#           # grep, write_file, ...) are omitted — with sandbox:execute denied
+#           # they could only ever return the deny error, so allow just the
+#           # non-sandbox tools.
+#           tools: {allow: ["web_search"]}
 #           routes: {allow: ["threads:read", "runs:read"]}
 #           models: {allow: ["gpt-4o-mini"]}
+#           sandbox: {allow: false}  # deny sandbox execution
 authorization:
   enabled: false
 

docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md

@@ -349,6 +349,60 @@ Phase 1 最低验证要求：
 - **延期：** Skills、Sandbox 权限（Phase 3 后续 PR）；前端 effective-permissions 展示；
   management route 的 provider 迁移。
 
+### 2026-08-02 — Phase 3 / Sandbox authorization (execute)
+
+- **背景：** Phase 3 Models 合并后，sandbox 仍只由 config presence
+  （`feat.sandbox is not False`）控制，任何已认证用户都能获得完整 sandbox 执行
+  （bash、文件 I/O）。RFC §9 要求 `SandboxMiddleware gates on
+  authorize("sandbox","execute")`，deny 时返回友好错误消息而非崩溃。
+- **决策（gate 位置）：** 与 Models/Skills 不同，sandbox 不是具名资源，而是一个
+  **执行环境** —— 多个工具（bash、read_file、write_file、glob、grep 等）都依赖它，
+  全部经过 `ensure_sandbox_initialized` / `ensure_sandbox_initialized_async`
+  （sandbox/tools.py）。选择在 **sandbox 获取的唯一入口** gate，而非在 middleware 里
+  维护"sandbox 工具名集合"（Shotgun Surgery，每加一个 sandbox 工具都要改 middleware）。
+  具体在两个 acquire 点之前调用共享的 `authorize_sandbox_execution` helper：
+  - lazy 路径：`ensure_sandbox_initialized` + async（覆盖所有 sandbox 工具）
+  - eager 路径：`SandboxMiddleware.before_agent` / `abefore_agent`（`lazy_init=False`）
+- **决策（授权语义）：** `authorize("sandbox", "execute", target="*")` —— **二元判断**
+  （"这个角色能否用 sandbox"），target 用 `"*"` 表示"sandbox 资源整体"。RBAC
+  `allow: ["*"]` / `allow: true` 允许，`allow: []` / `allow: false` 拒绝。
+- **决策（deny 行为）：** 新增 `SandboxAuthorizationError(SandboxError)`，deny 时抛出，
+  沿工具执行链传播 → agent 的 tool-error 处理转成友好 `ToolMessage`
+  （"sandbox execution is not permitted for your role"），符合 RFC §9 的"not a crash"。
+- **否决方案：** 不在 `SandboxMiddleware.wrap_tool_call` 里 per-tool gate —— middleware
+  无法区分哪些工具需要 sandbox，要么误伤非 sandbox 工具，要么维护硬编码工具名集合。
+  不为 sandbox 引入独立的 `SandboxMiddleware` 构造参数接收 provider —— lazy 路径不经过
+  middleware，gate 放在工具侧的 `ensure_sandbox_initialized` 才是 single source of truth。
+- **决策（Gateway 辅助同步路径）：** 多路径覆盖自审（pr-review 检查点 14）发现 4 个
+  绕过 `ensure_sandbox_initialized` 的直接 acquire：uploads.py（上传文件同步进 sandbox）、
+  artifacts.py（artifact 编辑后同步）、feishu.py / dingtalk.py（IM 下载文件同步）。
+  - uploads / artifacts（Gateway 路由，身份齐全）：加共享 helper `try_acquire_sandbox_for_request`（内部经 `authorize_sandbox_for_request` gate）
+    （gateway/authz.py，从 `request.state.user` 构造 Principal，含 INTERNAL_SYSTEM_ROLE pop）。
+    deny 时**跳过 sandbox 同步**（上传/artifact 编辑本身仍成功——deny 的 role 反正无法
+    通过 sandbox 消费这些文件）；provider 解析失败按 `fail_closed` 降级，不让 route 500。
+  - feishu / dingtalk（channel worker 路径）：**本 PR 不 gate**。理由：channel 文件下载路径
+    无法拿到完整授权身份（owner-user 解析依赖 run 启动时的 `inject_authenticated_user_context`，
+    文件下载时不可得）；且 deny 时同步的文件无法被 agent 消费，仅浪费一次幂等 acquire。
+    留作 follow-up（若维护者要求，可从 channel worker 的 run context 传递身份）。
+- **兼容性：** `authorization.enabled: false` 时 `authorize_sandbox_execution` 是 no-op
+  （直接返回）。RBAC provider 的 `_RESOURCE_POLICY_KEYS` 已包含 `"sandbox": "sandbox"`，
+  `provider.py` 已声明 `"sandbox"` 为有效 resource，无需 schema 变更。对 test mock
+  （SimpleNamespace app_config）安全：使用 `getattr` + `is not True` 防御。
+- **证据：** `tests/test_sandbox_authorization.py`（23 tests）覆盖 disabled/RBAC allow/deny/
+  deny-via-bool/no-policy-unrestricted/provider-error-fail-closed-fail-open/
+  internal-caller/default-role 场景，deny 错误携带 role，provider 收到正确的
+  resource/action/target；`ensure_sandbox_initialized` sync+async 的 deny（不 acquire）
+  + allow（acquire）集成场景；eager 路径（`before_agent` + `abefore_agent`）deny 跳过
+  acquire 而非 run 级报错；provider 解析错误的 fail-closed/fail-open（含 fail-open 语义
+  反转回归）；uploads/artifacts 路由 deny（acquire 不被调用、主操作仍成功）+ allow
+  （acquire 被调用）集成场景；request=None 容忍（直调测试路径）；无 config.yaml 时
+  gate no-op（CI 环境）；mock app_config（SimpleNamespace）防御回归；既有 `test_sandbox_middleware.py`（22 tests）、
+  `test_artifacts_router.py`、`test_uploads_manager.py` 全部通过
+  （authorization 禁用时 gate 是 no-op，不破坏现有行为）。
+- **延期：** Phase 3（Models/Skills/Sandbox）三资源类型完成；Phase 4 前端
+  effective-permissions 展示；management route 的 provider 迁移；
+  feishu/dingtalk 文件同步路径的 sandbox gate（身份传递机制待定）。
+
 ### 新记录模板
 
 ```markdown