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

README.md

@@ -897,7 +897,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 @@ async def get_thread(thread_id: str, request: Request):
 
 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,
+    }
+
+
+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
 
-    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
+    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

@@ -15,10 +15,11 @@
 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
@@ -438,7 +439,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()
@@ -464,9 +472,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 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

@@ -22,6 +22,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.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 @@
     "Principal",
     "RbacAuthorizationProvider",
     "apply_tool_authorization",
+    "authorize_sandbox_execution",
     "build_principal_from_context",
     "filter_tools_by_authorization",
     "normalize_authz_attributes",