fix(sandbox): enforce execution authorization for image reads (#5799)

* fix(sandbox): enforce execution authorization for image reads

* refactor(sandbox): share image injection authorization plan

* docs(agents): trim image middleware guidance to fit CI budget

---------

Co-authored-by: YxinMiracle <“939157765@qq.com”>

README.md

@@ -1275,6 +1275,8 @@ Each message can still select up to 1000 documents. When more than 100 documents
 
 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. Every HTTP route that starts or enables a future Agent run requires `runs:create`: this includes the stateless `POST /api/runs/stream` and `POST /api/runs/wait` endpoints plus scheduled-task create, update, resume, and manual-trigger mutations. Scheduled-task mutations retain their existing `threads:write` requirement, and the stateless routes separately enforce ownership when the optional thread ID is supplied in the request body. 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).
 
+For vision-capable agents, `view_image` and the subsequent model-context image read also require `sandbox:execute`. Allowing the tool name alone does not grant image-file access; a role denied sandbox execution cannot reread previously recorded image metadata after its permissions change.
+
 Advanced deployments can also extend the agent runtime itself by declaring `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. Each entry is a `module.path:ClassName` string (zero-argument constructor) or an object `{class, kwargs}` whose `kwargs` are passed to the constructor. `kwargs` values must be JSON types (object, array, string, number, boolean, or null); YAML dates and timestamps are coerced to ISO strings so they match JSON. DeerFlow loads the same configured 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, broken modules, and constructor errors 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`. Separate lead-only/subagent-only middleware lists are not supported yet.
 
 For packaged and configurable runtime integrations, use DeerFlow's extension manager.

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

@@ -109,7 +109,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
 21. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is read from terminal `ToolMessage.additional_kwargs` in the current run and merged back into the dispatching AIMessage by message position. The same state update marks the ToolMessage with `subagent_token_usage_attributed=true`, so checkpoint replay or middleware re-entry cannot add the cumulative snapshot twice; missing/malformed usage or a result with no matching dispatch remains unmarked and retryable.
 22. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint.
 23. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses); captures the runtime-resolved user so standalone LangGraph Server reads and writes stay in the same bucket
-24. **ViewImageMiddleware** - *(optional, if the model supports vision)* Appends a hidden HumanMessage with base64 image data, identified by a reserved ID prefix plus a server-owned metadata marker, to `ModelRequest.messages` in `wrap_model_call` / `awrap_model_call`. The payload lives only in that request and is never returned as a state update, so no checkpoint carries it and an interrupted run cannot strand it in history; state keeps only the lightweight `viewed_images` metadata. It owns that context and rebuilds it per call: its own message is swept out of the request first — a thread checkpointed by the earlier `before_model`/`after_model` pair (which wrote the payload into state and took it back out with `RemoveMessage`) can carry one that reached state but was never removed, and leaving it in would resend that base64 in every later request for the life of the thread — then a freshly built one is appended when warranted. The sweep requires both the reserved ID prefix and the server-owned marker, so a client cannot get its own message dropped; unmarked leftovers predating the marker are left in place and merely not duplicated
+24. **ViewImageMiddleware** - *(vision models only)* Rebuilds hidden base64 image context in `ModelRequest.messages`, never in new checkpoint state; state retains lightweight `viewed_images` metadata. `_image_injection_plan` first sweeps stale server-marked context (including payloads stranded by the old `before_model`/`after_model` lifecycle), then shares one read-eligibility predicate between sync injection and async authorization. The sweep requires both the reserved ID prefix and server-owned marker, preserving user messages and unmarked legacy payloads. Before reading images, both paths recheck `sandbox:execute`, including restored views after role changes; denial skips I/O and removes marked stale context. Only the async wrapper passes `authorization_checked=True` after its gate, avoiding a duplicate synchronous provider call in its worker.
 25. **McpRoutingMiddleware** - Auto-promotes deferred schemas matching the latest real user message before `DeferredToolFilterMiddleware`, without executing tools. New names emit `middleware:tool_promotion` with `source=routing_hint`; repeat passes emit nothing
 26. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, derives subagent attribution from the server-installed recorder, returns the original `Command`, and omits private payloads
 27. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)

backend/packages/harness/deerflow/agents/middlewares/view_image_middleware.py

@@ -3,7 +3,7 @@
 import base64
 import hashlib
 import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Mapping
 from pathlib import Path
 from typing import override
 from uuid import uuid4
@@ -365,11 +365,26 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
             },
         )
 
-    def _inject(self, request: ModelRequest) -> ModelRequest:
+    @staticmethod
+    def _authorization_context(request: ModelRequest) -> Mapping:
+        context = getattr(request.runtime, "context", None)
+        return context if isinstance(context, Mapping) else {}
+
+    def _image_injection_plan(self, request: ModelRequest) -> tuple[list[AnyMessage], bool, bool]:
+        """Share message cleanup and read eligibility across sync/async paths."""
+        messages = [message for message in request.messages if not self._is_image_context_message(message)]
+        should_inject = self._should_inject_image_message(messages)
+        needs_authorization = should_inject and bool((request.state or {}).get("viewed_images"))
+        return messages, should_inject, needs_authorization
+
+    def _inject(self, request: ModelRequest, *, authorization_checked: bool = False) -> ModelRequest:
         """Rebuild the request's image context from ``viewed_images``.
 
         Args:
             request: The pending model request
+            authorization_checked: True only after ``awrap_model_call`` has
+                enforced ``sandbox:execute`` for this exact request; skips the
+                synchronous recheck inside its worker thread.
 
         Returns:
             A request whose messages carry exactly the image context this call
@@ -384,14 +399,25 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
         # request for the life of the thread. Matching requires both the reserved
         # ID prefix and the server-owned marker, and Gateway strips that marker
         # from client input, so this can never drop a user-authored message.
-        messages = [message for message in request.messages if not self._is_image_context_message(message)]
+        messages, should_inject, needs_authorization = self._image_injection_plan(request)
         dropped_stranded = len(messages) != len(request.messages)
         if dropped_stranded:
             logger.debug("Dropping %d stranded image context message(s) from the model request", len(request.messages) - len(messages))
 
-        if not self._should_inject_image_message(messages):
+        if not should_inject:
             return request.override(messages=messages) if dropped_stranded else request
 
+        if needs_authorization and not authorization_checked:
+            from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
+            from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+            try:
+                authorize_sandbox_execution(context=self._authorization_context(request), app_config=safe_app_config())
+            except SandboxAuthorizationError:
+                # A restored view may predate a role/policy change. Never read
+                # its host mirror or remote sandbox for an unauthorized request.
+                return request.override(messages=messages) if dropped_stranded else request
+
         # Mixed content (text + images) for the model only, so hide it from the
         # chat UI and IM channels (matches the other middleware-injected context
         # messages) even though it never leaves this request.
@@ -421,5 +447,17 @@ class ViewImageMiddleware(AgentMiddleware[ViewImageMiddlewareState]):
         # blocking work without allowing cancellation to outlive a sandbox
         # client operation. The outer run lease may release the client as soon as
         # cancellation propagates.
-        injected_request = await run_sync_lifecycle_operation(self._inject, request)
+        messages, _, needs_authorization = self._image_injection_plan(request)
+        if needs_authorization:
+            from deerflow.authz.sandbox_authz import authorize_sandbox_execution_async, safe_app_config_async
+            from deerflow.sandbox.exceptions import SandboxAuthorizationError
+
+            try:
+                await authorize_sandbox_execution_async(context=self._authorization_context(request), app_config=await safe_app_config_async())
+            except SandboxAuthorizationError:
+                # Still sweep any old model-only image context before handing
+                # the request on; do not schedule a read in a worker.
+                return await handler(request.override(messages=messages))
+
+        injected_request = await run_sync_lifecycle_operation(self._inject, request, authorization_checked=True)
         return await handler(injected_request)

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

@@ -23,7 +23,7 @@ Only standalone tool discovery without a model falls back to the base profile.
 3. **Built-in tools**:
    - `present_files` - Make output files visible to user (only `/mnt/user-data/outputs`); virtual paths use `resolve_runtime_user_id(runtime)` so validation resolves the same user-scoped outputs directory established by `ThreadDataMiddleware`
    - `ask_clarification` - Request clarification (intercepted by ClarificationMiddleware, which preserves text fallback and adds `artifact.human_input` for Web UI Human Input Cards). Beyond free text and single choice, the request-side v2 protocol supports `fields` (structured form card collecting several values at once; field types: text/textarea/number/select/multi_select/checkbox/date, validated and normalized server-side in the middleware — invalid entries are dropped, unknown types degrade to `text`; a standalone multi-select question is a one-field form). Replies stay on the v1 response protocol (`text`/`option`): the form card submits a readable text summary
-   - `view_image` - Read image bytes for vision-capable models; live sandbox bytes win for the same sandbox generation, replacement-sandbox recovery uses only SHA-256-verified synchronized host bytes, and async tool invocation drains blocking reads before cancellation may release the sandbox lease
+   - `view_image` - Read image bytes for vision-capable models; both sync and async entry points require `sandbox:execute` before any host or sandbox read. Live sandbox bytes win for the same sandbox generation, replacement-sandbox recovery uses only SHA-256-verified synchronized host bytes, and async tool invocation drains blocking reads before cancellation may release the sandbox lease
    - `setup_agent` - Bootstrap-only: persist a custom agent's `SOUL.md` and `config.yaml`. Re-bootstrapping preserves the owner's existing `display_name`. Bound only when `is_bootstrap=True`.
    - `update_agent` - Custom-agent-only: persist self-updates to the current agent's `SOUL.md` / `config.yaml` from inside a normal chat (partial update + atomic write). Bound when `agent_name` is set and `is_bootstrap=False`.
 4. **Subagent tool** (if enabled):

backend/packages/harness/deerflow/tools/builtins/view_image_tool.py

@@ -122,6 +122,14 @@ def _view_image(
     Args:
         image_path: Absolute /mnt/user-data virtual path to the image file. Common formats supported: jpg, jpeg, png, webp, gif.
     """
+    from deerflow.sandbox.tools import sandbox_authorization_scope
+
+    with sandbox_authorization_scope(runtime):
+        return _view_image_authorized(runtime, image_path, tool_call_id)
+
+
+def _view_image_authorized(runtime: Runtime, image_path: str, tool_call_id: str) -> Command:
+    """Read image bytes only after the sandbox execution gate has passed."""
     from deerflow.sandbox.exceptions import SandboxRuntimeError
     from deerflow.sandbox.overwrite import unwrap_sandbox
     from deerflow.sandbox.sandbox_provider import get_sandbox_provider
@@ -288,13 +296,15 @@ async def _aview_image(
 ) -> Command:
     """Run the blocking image read without letting cancellation outlive it."""
     from deerflow.sandbox.lease import run_sync_lifecycle_operation
+    from deerflow.sandbox.tools import sandbox_authorization_scope_async
 
-    return await run_sync_lifecycle_operation(
-        _view_image,
-        runtime,
-        image_path,
-        tool_call_id,
-    )
+    async with sandbox_authorization_scope_async(runtime):
+        return await run_sync_lifecycle_operation(
+            _view_image_authorized,
+            runtime,
+            image_path,
+            tool_call_id,
+        )
 
 
 view_image_tool = StructuredTool.from_function(

backend/tests/test_view_image_sandbox_authorization_regression.py

@@ -0,0 +1,219 @@
+"""Offline regression and CLI probe for view_image sandbox authorization.
+
+From ``backend/``::
+
+    .venv/bin/python tests/test_view_image_sandbox_authorization_regression.py --expect vulnerable
+    .venv/bin/python tests/test_view_image_sandbox_authorization_regression.py --expect blocked
+    .venv/bin/python -m pytest tests/test_view_image_sandbox_authorization_regression.py -q
+
+The probe uses one synthetic GIF in a temporary directory. It needs no server,
+model provider, credentials, or network access.
+"""
+
+import argparse
+import base64
+import json
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from langchain.agents.middleware.types import ModelRequest
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+
+from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
+from deerflow.authz.adapter import GuardrailAuthorizationAdapter
+from deerflow.authz.rbac import RbacAuthorizationProvider
+from deerflow.authz.sandbox_authz import authorize_sandbox_execution
+from deerflow.authz.tool_filter import apply_tool_authorization
+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.guardrails.provider import GuardrailRequest
+from deerflow.sandbox.exceptions import SandboxAuthorizationError
+from deerflow.tools.builtins.view_image_tool import view_image_tool
+
+GIF_BYTES = base64.b64decode("R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7")
+IMAGE_PATH = "/mnt/user-data/uploads/synthetic.gif"
+
+
+def _setup(root: Path, monkeypatch: pytest.MonkeyPatch, *, sandbox_allowed: bool):
+    uploads = root / "uploads"
+    uploads.mkdir()
+    image_file = uploads / "synthetic.gif"
+    image_file.write_bytes(GIF_BYTES)
+    context = {"thread_id": "synthetic-thread", "user_id": "synthetic-user", "user_role": "reviewer"}
+    runtime = SimpleNamespace(
+        state={"thread_data": {"workspace_path": str(root / "workspace"), "uploads_path": str(uploads), "outputs_path": str(root / "outputs")}},
+        context=context,
+        config={},
+    )
+    config = AppConfig(
+        models=[ModelConfig(name="synthetic", model="synthetic", use="langchain_openai:ChatOpenAI", supports_vision=True)],
+        sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
+        authorization=AuthorizationConfig(enabled=True, fail_closed=True, default_role="reviewer"),
+    )
+    provider = RbacAuthorizationProvider(roles={"reviewer": {"tools": {"allow": ["view_image"]}, "sandbox": {"allow": sandbox_allowed}}})
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
+    monkeypatch.setattr("deerflow.authz.sandbox_authz.resolve_authorization_provider", lambda _: provider)
+    authorized_tools, _ = apply_tool_authorization([view_image_tool], context=context, app_config=config, authorization_provider=provider)
+    assert [tool.name for tool in authorized_tools] == ["view_image"]
+    tool_decision = GuardrailAuthorizationAdapter(provider, default_role="reviewer").evaluate(
+        GuardrailRequest(tool_name="view_image", tool_input={"image_path": IMAGE_PATH}, thread_id=context["thread_id"], user_id=context["user_id"], user_role=context["user_role"])
+    )
+    assert tool_decision.allow
+    return runtime, config, image_file
+
+
+def _model_request(runtime, viewed_images: dict, tool_message: ToolMessage | None = None) -> ModelRequest:
+    messages = [
+        HumanMessage(content="Describe this synthetic image"),
+        AIMessage(content="", tool_calls=[{"name": "view_image", "args": {"image_path": IMAGE_PATH}, "id": "image-call"}]),
+        tool_message or ToolMessage(content="Successfully read image", tool_call_id="image-call"),
+    ]
+    return ModelRequest(
+        model=FakeMessagesListChatModel(responses=[AIMessage(content="ok")]),
+        messages=messages,
+        system_message=None,
+        tool_choice=None,
+        tools=[],
+        response_format=None,
+        state={"messages": messages, "viewed_images": viewed_images},
+        runtime=SimpleNamespace(context=runtime.context),
+        model_settings={},
+    )
+
+
+def _image_payloads(request: ModelRequest) -> list[dict]:
+    return [block for message in request.messages for block in (message.content if isinstance(message.content, list) else []) if isinstance(block, dict) and block.get("type") == "image_url"]
+
+
+def _image_metadata(image_file: Path) -> dict:
+    return {IMAGE_PATH: {"actual_path": str(image_file), "mime_type": "image/gif", "size": len(GIF_BYTES)}}
+
+
+def test_allowed_sync_read_still_reaches_model(tmp_path, monkeypatch):
+    runtime, config, _ = _setup(tmp_path, monkeypatch, sandbox_allowed=True)
+    authorize_sandbox_execution(context=runtime.context, app_config=config)
+    result = view_image_tool.func(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+    assert result.update["messages"][0].content == "Successfully read image"
+    request = ViewImageMiddleware()._inject(_model_request(runtime, result.update["viewed_images"], result.update["messages"][0]))
+    assert _image_payloads(request)[0]["image_url"]["url"] == f"data:image/gif;base64,{base64.b64encode(GIF_BYTES).decode()}"
+
+
+def test_denied_sync_tool_stops_before_image_read(tmp_path, monkeypatch):
+    runtime, config, image_file = _setup(tmp_path, monkeypatch, sandbox_allowed=False)
+    with pytest.raises(SandboxAuthorizationError):
+        authorize_sandbox_execution(context=runtime.context, app_config=config)
+    original_open = open
+    reads = []
+
+    def record_open(file, *args, **kwargs):
+        if isinstance(file, (str, bytes, Path)) and Path(file).resolve() == image_file.resolve():
+            reads.append(file)
+        return original_open(file, *args, **kwargs)
+
+    with patch("builtins.open", side_effect=record_open):
+        with pytest.raises(SandboxAuthorizationError):
+            view_image_tool.func(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+    assert reads == []
+
+
+def test_denied_restored_view_does_not_reenter_model(tmp_path, monkeypatch):
+    runtime, _, _ = _setup(tmp_path, monkeypatch, sandbox_allowed=True)
+    viewed = view_image_tool.func(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call").update["viewed_images"]
+    denied_provider = RbacAuthorizationProvider(roles={"reviewer": {"tools": {"allow": ["view_image"]}, "sandbox": {"allow": False}}})
+    monkeypatch.setattr("deerflow.authz.sandbox_authz.resolve_authorization_provider", lambda _: denied_provider)
+    stranded = ViewImageMiddleware._create_image_context_message([{"type": "image_url", "image_url": {"url": "data:image/gif;base64,stale"}}])
+    original_request = _model_request(runtime, viewed)
+    original_request = original_request.override(messages=[*original_request.messages, stranded])
+    with patch("builtins.open", side_effect=AssertionError("denied image read")):
+        request = ViewImageMiddleware()._inject(original_request)
+    assert _image_payloads(request) == []
+    assert stranded not in request.messages
+
+
+def test_denied_tool_does_not_touch_live_sandbox(tmp_path, monkeypatch):
+    runtime, _, _ = _setup(tmp_path, monkeypatch, sandbox_allowed=False)
+    runtime.state["sandbox"] = {"sandbox_id": "synthetic-remote"}
+    with patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", side_effect=AssertionError("denied sandbox lookup")):
+        with pytest.raises(SandboxAuthorizationError):
+            view_image_tool.func(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+
+
+@pytest.mark.asyncio
+async def test_denied_async_tool_and_restored_view_do_not_read(tmp_path, monkeypatch):
+    runtime, _, image_file = _setup(tmp_path, monkeypatch, sandbox_allowed=False)
+    stale_context = ViewImageMiddleware._create_image_context_message([{"type": "image_url", "image_url": {"url": "data:image/gif;base64,stale"}}])
+    user_message = HumanMessage(id="view-image-context:user-authored", content="Keep this message")
+    original_request = _model_request(runtime, _image_metadata(image_file))
+    original_request = original_request.override(messages=[*original_request.messages, stale_context, user_message])
+    with patch("builtins.open", side_effect=AssertionError("denied image read")):
+        with pytest.raises(SandboxAuthorizationError):
+            await view_image_tool.coroutine(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+
+        async def handler(request):
+            return request
+
+        request = await ViewImageMiddleware().awrap_model_call(original_request, handler)
+    assert _image_payloads(request) == []
+    assert stale_context not in request.messages
+    assert user_message in request.messages
+
+
+@pytest.mark.asyncio
+async def test_allowed_async_tool_and_model_injection(tmp_path, monkeypatch):
+    runtime, _, _ = _setup(tmp_path, monkeypatch, sandbox_allowed=True)
+    result = await view_image_tool.coroutine(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+
+    async def handler(request):
+        return request
+
+    request = await ViewImageMiddleware().awrap_model_call(_model_request(runtime, result.update["viewed_images"], result.update["messages"][0]), handler)
+    assert len(_image_payloads(request)) == 1
+
+
+def _run_probe() -> dict[str, object]:
+    with tempfile.TemporaryDirectory(prefix="deerflow-image-authz-") as temp_dir, pytest.MonkeyPatch.context() as monkeypatch:
+        runtime, config, image_file = _setup(Path(temp_dir), monkeypatch, sandbox_allowed=False)
+        with pytest.raises(SandboxAuthorizationError):
+            authorize_sandbox_execution(context=runtime.context, app_config=config)
+        original_open = open
+        image_reads = 0
+
+        def record_open(file, *args, **kwargs):
+            nonlocal image_reads
+            if isinstance(file, (str, bytes, Path)) and Path(file).resolve() == image_file.resolve():
+                image_reads += 1
+            return original_open(file, *args, **kwargs)
+
+        with patch("builtins.open", side_effect=record_open):
+            try:
+                result = view_image_tool.func(runtime=runtime, image_path=IMAGE_PATH, tool_call_id="image-call")
+            except SandboxAuthorizationError:
+                tool_denied = True
+                viewed_images = {}
+            else:
+                tool_denied = False
+                viewed_images = result.update.get("viewed_images", {})
+            request = ViewImageMiddleware()._inject(_model_request(runtime, viewed_images or _image_metadata(image_file)))
+        return {"tool_denied": tool_denied, "sandbox_denied": True, "image_reads": image_reads, "model_received_image": bool(_image_payloads(request))}
+
+
+def _main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--expect", choices=("vulnerable", "blocked"), required=True)
+    args = parser.parse_args()
+    observed = _run_probe()
+    vulnerable = not observed["tool_denied"] and observed["image_reads"] >= 2 and observed["model_received_image"]
+    blocked = observed["tool_denied"] and observed["image_reads"] == 0 and not observed["model_received_image"]
+    status = "vulnerable" if vulnerable else "blocked" if blocked else "inconclusive"
+    print(json.dumps({**observed, "observed": status, "expected": args.expect}, sort_keys=True))
+    return 0 if status == args.expect else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(_main())