Dashboard file access security hardening

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #57660 fix(security): salvage six P1 hardening PRs — browser guards, MEDIA anchoring, .env lockdown, delegate ACP transport, matrix sync isolation

  merged to main

  files: gateway/platforms/api_server.py, hermes_cli/web_server.py, plugins/platforms/matrix/adapter.py, run_agent.py, tests/gateway/test_api_server_media_data_urls.py, tests/gateway/test_matrix.py, tests/hermes_cli/test_web_server_files.py, tests/tools/test_browser_camofox_private_page_guard.py, tests/tools/test_browser_cdp_tool.py, tests/tools/test_delegate.py, tools/browser_camofox.py, tools/browser_cdp_tool.py, tools/delegate_tool.py

  Show the diff

  • gateway/platforms/api_server.py

    @@ -54,9 +54,11 @@
     
     from gateway.config import Platform, PlatformConfig
     from gateway.platforms.base import (
    +    MEDIA_TAG_CLEANUP_RE,
         BasePlatformAdapter,
         SendResult,
         is_network_accessible,
    +    validate_media_delivery_path,
     )
     from agent.redact import redact_sensitive_text
     
    @@ -581,9 +583,6 @@ async def cors_middleware(request, handler):
         ".webp": "image/webp",
         ".bmp": "image/bmp",
     }
    -_MEDIA_TAG_RE = re.compile(
    -    r"[`\"']?MEDIA:\s*(`[^`\n]+`|\"[^\"\n]+\"|'[^'\n]+'|\S+)[`\"']?"
    -)
     _MEDIA_DATA_URL_MAX_BYTES = 5 * 1024 * 1024  # skip images larger than 5MB
     
     
    @@ -594,18 +593,35 @@ def _resolve_media_to_data_urls(text: str) -> str:
         ``MEDIA:`` tags referencing images on the server are useless to them.
         Inline small local images as markdown data URLs; non-image or unreadable
         paths are left untouched.
    +
    +    Uses the same anchored ``MEDIA_TAG_CLEANUP_RE`` matcher and
    +    ``validate_media_delivery_path`` safety check every other platform
    +    adapter's media delivery already goes through (gateway/platforms/base.py)
    +    — an absolute-path anchor plus a known-extension requirement, and a
    +    resolved-path check against the credential/system-path denylist. The
    +    prior pattern here matched any bare token after ``MEDIA:`` (including a
    +    relative/traversal path like ``../../etc/passwd.png``) and read the file
    +    directly with no denylist, so any image-suffixed, readable file the
    +    process could see was base64-exfiltrated to the API caller if its path
    +    merely appeared in the model's own final reply text.
         """
         if not text or "MEDIA:" not in text:
             return text
         import base64
     
         def _to_data_url(path_str: str) -> Optional[str]:
    -        p = Path(path_str.strip().strip("`\"'")).expanduser()
    +        # validate_media_delivery_path() strips wrapping quotes/backticks
    +        # and trailing punctuation internally, same as MEDIA_TAG_CLEANUP_RE's
    +        # other callers (extract_media / _strip_media_tag_directives) rely on.
    +        safe_path = validate_media_delivery_path(path_str)
    +        if not safe_path:
    +            return None
    +        p = Path(safe_path)
             suffix = p.suffix.lower()
             if suffix not in _MEDIA_IMG_EXT:
                 return None
             try:
    -            if not p.is_file() or p.stat().st_size > _MEDIA_DATA_URL_MAX_BYTES:
    +            if p.stat().st_size > _MEDIA_DATA_URL_MAX_BYTES:
                     return None
                 b64 = base64.b64encode(p.read_bytes()).decode()
             except OSError:
    @@ -613,10 +629,10 @@ def _to_data_url(path_str: str) -> Optional[str]:
             return f"![image](data:{_MEDIA_MIME[suffix]};base64,{b64})"
     
         def _repl(m: "re.Match[str]") -> str:
    -        return _to_data_url(m.group(1)) or m.group(0)
    +        return _to_data_url(m.group("path")) or m.group(0)
     
         try:
    -        return _MEDIA_TAG_RE.sub(_repl, text)
    +        return MEDIA_TAG_CLEANUP_RE.sub(_repl, text)
         except Exception:
             return text
     

  • hermes_cli/web_server.py

    @@ -1186,6 +1186,19 @@ class ManagedFilesPolicy:
         "target",
         "venv",
     }
    +
    +# Filenames that must never be listed, read, or downloaded through the
    +# managed-files API.  These typically contain credentials (API keys, tokens)
    +# and exposing them through the dashboard file browser is a security leak —
    +# see issue #57505.
    +def _is_sensitive_filename(name: str) -> bool:
    +    """Return True for ``.env`` and any ``.env.<suffix>`` variant.
    +
    +    Case-insensitive so ``.ENV`` / ``.Env.local`` on case-insensitive
    +    filesystems (macOS/Windows mounts) can't slip past the guard.
    +    """
    +    lowered = name.lower()
    +    return lowered == ".env" or lowered.startswith(".env.")
     _FS_DATA_URL_MAX_BYTES = 16 * 1024 * 1024
     _FS_TEXT_SOURCE_MAX_BYTES = 64 * 1024 * 1024
     _FS_TEXT_PREVIEW_MAX_BYTES = 512 * 1024
    @@ -1616,7 +1629,11 @@ async def list_managed_files(request: Request, path: Optional[str] = None):
             raise HTTPException(status_code=400, detail="Path is not a directory")
     
         try:
    -        entries = [_managed_file_entry(policy, child) for child in target.iterdir()]
    +        entries = [
    +            _managed_file_entry(policy, child)
    +            for child in target.iterdir()
    +            if not _is_sensitive_filename(child.name)
    +        ]
         except PermissionError:
             raise HTTPException(status_code=403, detail="Directory is not readable")
         except OSError as exc:
    @@ -1642,6 +1659,8 @@ async def read_managed_file(request: Request, path: str):
             raise HTTPException(status_code=404, detail="File not found")
         if not target.is_file():
             raise HTTPException(status_code=400, detail="Path is not a file")
    +    if _is_sensitive_filename(target.name):
    +        raise HTTPException(status_code=403, detail="Access to sensitive files is not allowed")
     
         try:
             size = target.stat().st_size
    @@ -1684,6 +1703,8 @@ async def download_managed_file(request: Request, path: str):
             raise HTTPException(status_code=404, detail="File not found")
         if not target.is_file():
             raise HTTPException(status_code=400, detail="Path is not a file")
    +    if _is_sensitive_filename(target.name):
    +        raise HTTPException(status_code=403, detail="Access to sensitive files is not allowed")
     
         try:
             size = target.stat().st_size

  • plugins/platforms/matrix/adapter.py

    @@ -2351,7 +2351,18 @@ async def _dispatch_sync(self, sync_data: Dict[str, Any]) -> None:
             if inspect.isawaitable(tasks):
                 tasks = await tasks
             if tasks:
    -            await asyncio.gather(*tasks)
    +            # return_exceptions=True so one failing event handler doesn't abort
    +            # the whole gather and silently drop the SIBLING events in the same
    +            # sync response (a bare gather re-raises the first exception, leaving
    +            # the rest of the batch unprocessed). Mirrors the invite/redaction
    +            # gathers above. Surface each failure instead of swallowing it.
    +            results = await asyncio.gather(*tasks, return_exceptions=True)
    +            for result in results:
    +                if isinstance(result, Exception):
    +                    logger.warning(
    +                        "Matrix: event handler failed during sync dispatch: %s",
    +                        result,
    +                    )
     
         def _is_self_sender(self, sender: str) -> bool:
             """Return True if the sender refers to the bot's own account.

  • run_agent.py

    @@ -5623,7 +5623,10 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             New DELEGATE_TASK_SCHEMA fields only need to be added here to reach all
             invocation paths (concurrent, sequential, inline).
             """
    -        from tools.delegate_tool import delegate_task as _delegate_task
    +        from tools.delegate_tool import (
    +            _strip_model_hidden_task_fields,
    +            delegate_task as _delegate_task,
    +        )
             # Delegations from the top-level MODEL always run in the background —
             # the model does not get to choose. delegate_task returns immediately
             # with a handle (one per task) and each subagent's result re-enters the
    @@ -5639,10 +5642,8 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             return _delegate_task(
                 goal=function_args.get("goal"),
                 context=function_args.get("context"),
    -            tasks=function_args.get("tasks"),
    +            tasks=_strip_model_hidden_task_fields(function_args.get("tasks")),
                 max_iterations=function_args.get("max_iterations"),
    -            acp_command=function_args.get("acp_command"),
    -            acp_args=function_args.get("acp_args"),
                 role=function_args.get("role"),
                 background=(not _is_subagent),
                 parent_agent=self,

  • tests/gateway/test_api_server_media_data_urls.py

    @@ -72,6 +72,40 @@ def test_multiple_tags(self):
             out = _resolve_media_to_data_urls(f"MEDIA:{p1}\nand MEDIA:{p2}")
             self.assertEqual(out.count("data:image/png;base64,"), 2)
     
    +    def test_relative_traversal_path_not_inlined(self):
    +        """A relative/traversal path must never be inlined — the anchored
    +        MEDIA_TAG_CLEANUP_RE matcher requires an absolute-path prefix
    +        (~/, /, or a Windows drive letter), so a bare relative token after
    +        MEDIA: is left as literal text rather than resolved against cwd."""
    +        text = "MEDIA:../../../../etc/passwd.png"
    +        self.assertEqual(_resolve_media_to_data_urls(text), text)
    +
    +    def test_credential_path_not_inlined_even_with_image_extension(self):
    +        """An absolute path under the credential/system-path denylist
    +        (validate_media_delivery_path) must not be inlined even though it
    +        has an allowed image extension and the tag matcher's shape."""
    +        text = "MEDIA:~/.ssh/id_rsa.png"
    +        self.assertEqual(_resolve_media_to_data_urls(text), text)
    +
    +    def test_symlink_escaping_to_denylisted_target_not_inlined(self):
    +        """A symlink whose resolved target lands under a denylisted system
    +        prefix (/etc) must not be inlined — validate_media_delivery_path
    +        resolves symlinks before the containment/denylist check runs, so
    +        the traversal can't be laundered through an innocuous-looking
    +        image-suffixed symlink name."""
    +        import os
    +        import tempfile
    +        from pathlib import Path
    +
    +        d = Path(tempfile.mkdtemp(prefix="hermes_media_test_symlink"))
    +        link = d / "shot.png"
    +        try:
    +            os.symlink("/etc/hosts", link)
    +        except OSError:
    +            self.skipTest("symlink creation not supported in this environment")
    +        text = f"MEDIA:{link}"
    +        self.assertEqual(_resolve_media_to_data_urls(text), text)
    +
     
     if __name__ == "__main__":
         unittest.main()

  • tests/gateway/test_matrix.py

    @@ -5276,3 +5276,36 @@ async def test_flag_clears_when_second_connect_resolves_device_id(self):
             assert None not in _verify_call.args[0]["@bot:example.org"]
     
             await adapter.disconnect()
    +
    +
    +class TestMatrixDispatchSyncIsolation:
    +    """A failing mautrix event handler must not abort the whole sync batch.
    +
    +    ``_dispatch_sync`` gathers the per-event handler tasks. Without
    +    ``return_exceptions=True`` the first exception aborts the gather and the
    +    sibling events in the same sync response are silently dropped.
    +    """
    +
    +    @pytest.mark.asyncio
    +    async def test_dispatch_sync_isolates_failing_handler(self, caplog):
    +        import logging
    +
    +        adapter = _make_adapter()
    +        ran = {"ok": False}
    +
    +        async def _boom():
    +            raise RuntimeError("handler boom")
    +
    +        async def _ok():
    +            ran["ok"] = True
    +
    +        client = MagicMock()
    +        client.handle_sync = MagicMock(return_value=[_boom(), _ok()])
    +        adapter._client = client
    +
    +        with caplog.at_level(logging.WARNING):
    +            # Must not raise despite the failing handler.
    +            await adapter._dispatch_sync({"next_batch": "s1"})
    +
    +        assert ran["ok"] is True  # the sibling handler still ran
    +        assert "event handler failed" in caplog.text  # failure surfaced, not swallowed

  • tests/hermes_cli/test_web_server_files.py

    @@ -488,3 +488,75 @@ async def close(self):
         # ... and no .upload temp file was left behind.
         leftovers = [p.name for p in target.parent.iterdir() if ".upload" in p.name]
         assert leftovers == [], f"temp upload files leaked on cancellation: {leftovers}"
    +
    +
    +def test_sensitive_env_files_hidden_from_listing(forced_files_client):
    +    """Regression test for #57505: .env files must not appear in directory listings."""
    +    client, root = forced_files_client
    +
    +    # Create a regular file and .env variants including shorthand suffixes.
    +    root.mkdir(parents=True, exist_ok=True)
    +    regular = root / "config.txt"
    +    regular.write_text("safe content")
    +    env_file = root / ".env"
    +    env_file.write_text("SECRET_KEY=[redacted:secret-assignment]")
    +    env_local = root / ".env.local"
    +    env_local.write_text("LOCAL_SECRET=[redacted:secret-assignment]")
    +    env_prod = root / ".env.prod"
    +    env_prod.write_text("PROD_SECRET=[redacted:secret-assignment]")
    +
    +    listing = client.get("/api/files", params={"path": str(root)})
    +    assert listing.status_code == 200
    +    names = [e["name"] for e in listing.json()["entries"]]
    +    assert "config.txt" in names
    +    assert ".env" not in names
    +    assert ".env.local" not in names
    +    assert ".env.prod" not in names
    +
    +
    +def test_sensitive_env_files_blocked_read(forced_files_client):
    +    """Regression test for #57505: .env files must not be readable."""
    +    client, root = forced_files_client
    +
    +    root.mkdir(parents=True, exist_ok=True)
    +    env_file = root / ".env"
    +    env_file.write_text("SECRET_KEY=[redacted:secret-assignment]")
    +
    +    resp = client.get("/api/files/read", params={"path": str(env_file)})
    +    assert resp.status_code == 403
    +
    +
    +def test_sensitive_env_files_blocked_download(forced_files_client):
    +    """Regression test for #57505: .env files must not be downloadable."""
    +    client, root = forced_files_client
    +
    +    root.mkdir(parents=True, exist_ok=True)
    +    env_file = root / ".env"
    +    env_file.write_text("SECRET_KEY=[redacted:secret-assignment]")
    +
    +    resp = client.get("/api/files/download", params={"path": str(env_file)})
    +    assert resp.status_code == 403
    +
    +
    +def test_sensitive_env_suffix_variants_blocked(forced_files_client):
    +    """Regression: .env.<suffix> shorthand variants (e.g. .env.prod) must also be blocked."""
    +    client, root = forced_files_client
    +
    +    root.mkdir(parents=True, exist_ok=True)
    +    for suffix in ("prod", "dev", "staging.local", "ci"):
    +        p = root / f".env.{suffix}"
    +        p.write_text(f"SECRET_{suffix}=abc123")
    +        assert client.get("/api/files/read", params={"path": str(p)}).status_code == 403
    +        assert client.get("/api/files/download", params={"path": str(p)}).status_code == 403
    +
    +
    +def test_sensitive_env_case_insensitive_blocked(forced_files_client):
    +    """Regression: .ENV / .Env.local casings must be blocked too (case-insensitive FS mounts)."""
    +    client, root = forced_files_client
    +
    +    root.mkdir(parents=True, exist_ok=True)
    +    for name in (".ENV", ".Env.local", ".eNv.PROD"):
    +        p = root / name
    +        p.write_text("SECRET=[redacted:secret-assignment]")
    +        assert client.get("/api/files/read", params={"path": str(p)}).status_code == 403
    +        assert client.get("/api/files/download", params={"path": str(p)}).status_code == 403

  • tests/tools/test_browser_camofox_private_page_guard.py

    @@ -83,6 +83,34 @@ def fail_http(*_args, **_kwargs):
         assert action_phrase in out["error"]
     
     
    [redacted:email](
    +    ("tool_call", "action_phrase"),
    +    [
    +        (lambda: browser_camofox.camofox_click("@e1", task_id="t1"), "click"),
    +        (
    +            lambda: browser_camofox.camofox_type("@e1", "do-not-send-this", task_id="t1"),
    +            "type",
    +        ),
    +        (lambda: browser_camofox.camofox_press("Enter", task_id="t1"), "press"),
    +    ],
    +)
    +def test_private_page_blocks_camofox_input_actions(monkeypatch, _session, tool_call, action_phrase):
    +    _block_active(monkeypatch)
    +
    +    def fail_post(*_args, **_kwargs):
    +        raise AssertionError("Camofox action HTTP call should not run on a private page")
    +
    +    monkeypatch.setattr(browser_camofox, "_post", fail_post)
    +
    +    out = json.loads(tool_call())
    +
    +    assert out["success"] is False
    +    assert PRIVATE_URL in out["error"]
    +    assert "private or internal address" in out["error"]
    +    assert action_phrase in out["error"]
    +    assert "do-not-send-this" not in json.dumps(out)
    +
    +
     def test_snapshot_still_runs_when_page_is_public(monkeypatch, _session):
         _public_page(monkeypatch)
     
    @@ -98,6 +126,29 @@ def test_snapshot_still_runs_when_page_is_public(monkeypatch, _session):
         assert out["element_count"] == 1
     
     
    +def test_camofox_click_still_runs_when_page_is_public(monkeypatch, _session):
    +    _public_page(monkeypatch)
    +    calls = []
    +
    +    def fake_post(path, body=None, timeout=None):
    +        calls.append((path, body, timeout))
    +        return {"url": "https://example.test/"}
    +
    +    monkeypatch.setattr(browser_camofox, "_post", fake_post)
    +
    +    out = json.loads(browser_camofox.camofox_click("@e1", task_id="t1"))
    +
    +    assert out["success"] is True
    +    assert out["clicked"] == "e1"
    +    assert calls == [
    +        (
    +            "/tabs/tab-1/click",
    +            {"userId": "user-1", "ref": "e1"},
    +            None,
    +        )
    +    ]
    +
    +
     def test_guard_inactive_does_not_probe(monkeypatch, _session):
         """When the SSRF guard is inactive the read proceeds WITHOUT probing the URL.
     

  • tests/tools/test_browser_cdp_tool.py

    @@ -430,6 +430,70 @@ async def fake_call(*args, **kwargs):
         assert calls == []
     
     
    +def test_frame_id_route_blocked_when_current_page_is_private(monkeypatch):
    +    """frame_id routing (OOPIF via supervisor) must not bypass the guard
    +    applied to the stateless path — same private-page boundary either way."""
    +    supervisor_calls = []
    +
    +    import tools.browser_tool as bt
    +
    +    monkeypatch.setattr(bt, "_eval_ssrf_guard_active", lambda task_id: True)
    +    monkeypatch.setattr(bt, "_current_page_private_url", lambda task_id: PRIVATE_URL)
    +
    +    def fake_supervisor_route(**kwargs):
    +        supervisor_calls.append(kwargs)
    +        return json.dumps({"success": True, "result": {"value": "private data"}})
    +
    +    monkeypatch.setattr(
    +        browser_cdp_tool, "_browser_cdp_via_supervisor", fake_supervisor_route
    +    )
    +
    +    result = json.loads(
    +        browser_cdp_tool.browser_cdp(
    +            method="Runtime.evaluate",
    +            params={"expression": "document.body.innerText"},
    +            frame_id="frame-1",
    +            task_id="task-1",
    +        )
    +    )
    +
    +    assert "error" in result
    +    assert PRIVATE_URL in result["error"]
    +    assert "private or internal address" in result["error"]
    +    assert supervisor_calls == []
    +
    +
    +def test_frame_id_route_allowed_when_page_is_not_private(monkeypatch):
    +    """Sanity check: the new guard call must not block ordinary frame_id
    +    routing when the current page isn't private."""
    +    supervisor_calls = []
    +
    +    import tools.browser_tool as bt
    +
    +    monkeypatch.setattr(bt, "_eval_ssrf_guard_active", lambda task_id: True)
    +    monkeypatch.setattr(bt, "_current_page_private_url", lambda task_id: None)
    +
    +    def fake_supervisor_route(**kwargs):
    +        supervisor_calls.append(kwargs)
    +        return json.dumps({"success": True, "result": {"value": "ok"}})
    +
    +    monkeypatch.setattr(
    +        browser_cdp_tool, "_browser_cdp_via_supervisor", fake_supervisor_route
    +    )
    +
    +    result = json.loads(
    +        browser_cdp_tool.browser_cdp(
    +            method="Runtime.evaluate",
    +            params={"expression": "document.title"},
    +            frame_id="frame-1",
    +            task_id="task-1",
    +        )
    +    )
    +
    +    assert result.get("success") is True
    +    assert len(supervisor_calls) == 1
    +
    +
     def test_page_navigate_to_private_url_blocked_before_cdp(monkeypatch):
         calls = []
     

  • tests/tools/test_delegate.py

    @@ -78,6 +78,12 @@ def test_schema_valid(self):
             # config-authoritative via delegation.max_iterations so users get
             # predictable budgets.
             self.assertNotIn("max_iterations", props)
    +        # ACP subprocess transport is operator-controlled via config.yaml, not
    +        # model-controlled via delegate_task arguments.
    +        self.assertNotIn("acp_command", props)
    +        self.assertNotIn("acp_args", props)
    +        self.assertNotIn("acp_command", props["tasks"]["items"]["properties"])
    +        self.assertNotIn("acp_args", props["tasks"]["items"]["properties"])
             self.assertNotIn("maxItems", props["tasks"])  # removed — limit is now runtime-configurable
     
         def test_schema_description_advertises_runtime_limits(self):
    @@ -522,16 +528,7 @@ def test_build_child_agent_does_not_raise_name_error(self):
                     )
     
         def test_build_child_agent_ignores_acp_command_when_binary_missing(self):
    -        """Regression: _build_child_agent must not force provider='copilot-acp'
    -        when the override_acp_command binary is not on PATH.
    -
    -        Without this guard, a model that hallucinates
    -        ``delegate_task(acp_command="copilot")`` on a host without the Copilot
    -        CLI installed (Railway / headless containers / fresh VPS) would route
    -        the subagent through CopilotACPClient, which spawns the binary via
    -        subprocess and raises RuntimeError. After 3 retries the asyncio loop
    -        teardown can take the entire gateway down.
    -        """
    +        """Stale delegation.command config must not force ACP subprocess mode."""
             parent = _make_mock_parent(depth=0)
             # The crash scenario is a TG/cron agent on a host with no ACP CLI —
             # parent itself has no acp_command, so clearing the override must NOT
    @@ -606,68 +603,20 @@ def test_build_child_agent_honors_acp_command_when_binary_present(self):
             self.assertEqual(captured["provider"], "copilot-acp")
             self.assertEqual(captured["acp_command"], "copilot")
     
    -    def test_schema_prunes_acp_command_when_no_acp_binary(self):
    -        """Schema-level defense: delegate_task tool schema must NOT advertise
    -        acp_command / acp_args to the model when no ACP binary is installed.
    -
    -        Headless deploys (Railway / Fly / Docker / fresh VPS) typically have
    -        none of copilot / claude / codex. Without the schema prune, models
    -        occasionally hallucinate ``acp_command="copilot"`` from the field's
    -        description and crash subagent runs.
    -        """
    -        from tools.delegate_tool import _build_dynamic_schema_overrides
    -
    -        with patch("tools.delegate_tool._acp_binary_available", return_value=False):
    -            overrides = _build_dynamic_schema_overrides()
    -
    -        props = overrides["parameters"]["properties"]
    -        self.assertNotIn("acp_command", props, "top-level acp_command must be pruned")
    -        self.assertNotIn("acp_args", props, "top-level acp_args must be pruned")
    -
    -        task_item_props = props["tasks"]["items"]["properties"]
    -        self.assertNotIn(
    -            "acp_command", task_item_props, "per-task acp_command must be pruned"
    -        )
    -        self.assertNotIn(
    -            "acp_args", task_item_props, "per-task acp_args must be pruned"
    -        )
    -
    -    def test_schema_keeps_acp_command_when_binary_available(self):
    -        """Backward compat: when an ACP CLI IS on PATH, schema is unchanged.
    -        Users with working ACP setups must still be able to invoke it.
    -        """
    +    def test_schema_never_exposes_acp_transport_fields(self):
    +        """delegate_task must never make ACP transport model-facing."""
             from tools.delegate_tool import _build_dynamic_schema_overrides
     
    -        with patch("tools.delegate_tool._acp_binary_available", return_value=True):
    +        with patch("shutil.which", return_value="/usr/local/bin/copilot"):
                 overrides = _build_dynamic_schema_overrides()
     
             props = overrides["parameters"]["properties"]
    -        self.assertIn("acp_command", props)
    -        self.assertIn("acp_args", props)
    +        self.assertNotIn("acp_command", props)
    +        self.assertNotIn("acp_args", props)
     
             task_item_props = props["tasks"]["items"]["properties"]
    -        self.assertIn("acp_command", task_item_props)
    -        self.assertIn("acp_args", task_item_props)
    -
    -    def test_acp_binary_available_checks_known_clis(self):
    -        """_acp_binary_available must check the known ACP CLI names via
    -        shutil.which — guards against typos or accidental list trimming.
    -        """
    -        from tools.delegate_tool import _KNOWN_ACP_BINARIES, _acp_binary_available
    -
    -        self.assertIn("copilot", _KNOWN_ACP_BINARIES)
    -
    -        calls = []
    -
    -        def fake_which(name):
    -            calls.append(name)
    -            return None
    -
    -        with patch("shutil.which", side_effect=fake_which):
    -            self.assertFalse(_acp_binary_available())
    -
    -        for name in _KNOWN_ACP_BINARIES:
    -            self.assertIn(name, calls)
    +        self.assertNotIn("acp_command", task_item_props)
    +        self.assertNotIn("acp_args", task_item_props)
     
         def test_saved_tool_names_set_on_child_before_run(self):
             """_run_single_child must set _delegate_saved_tool_names on the child
    @@ -2281,37 +2230,39 @@ def test_invalid_reasoning_effort_falls_back_to_parent(self, MockAgent, mock_cfg
     class TestDispatchDelegateTask(unittest.TestCase):
         """Tests for the _dispatch_delegate_task helper and full param forwarding."""
     
    -    @patch("tools.delegate_tool._load_config", return_value={})
    -    @patch("tools.delegate_tool._resolve_delegation_credentials")
    -    def test_acp_args_forwarded(self, mock_creds, mock_cfg):
    -        """Both acp_command and acp_args reach delegate_task via the helper."""
    -        mock_creds.return_value = {
    -            "provider": None, "base_url": None,
    -            "api_key": None, "api_mode": None, "model": None,
    -        }
    -        parent = _make_mock_parent(depth=0)
    -        with patch("tools.delegate_tool._build_child_agent") as mock_build:
    -            mock_child = MagicMock()
    -            mock_child.run_conversation.return_value = {
    -                "final_response": "done", "completed": True,
    -                "api_calls": 1, "messages": [],
    -            }
    -            mock_child._delegate_saved_tool_names = []
    -            mock_child._credential_pool = None
    -            mock_child.session_prompt_tokens = 0
    -            mock_child.session_completion_tokens = 0
    -            mock_child.model = "test"
    -            mock_build.return_value = mock_child
    +    def test_model_acp_args_not_forwarded(self):
    +        """The live model dispatch path strips hidden ACP transport args."""
    +        import run_agent
     
    -            delegate_task(
    -                goal="test",
    -                acp_command="claude",
    -                acp_args=["--acp", "--stdio"],
    -                parent_agent=parent,
    +        captured = {}
    +
    +        def fake_delegate_task(**kwargs):
    +            captured.update(kwargs)
    +            return "{}"
    +
    +        parent = _make_mock_parent(depth=0)
    +        with patch("tools.delegate_tool.delegate_task", fake_delegate_task):
    +            run_agent.AIAgent._dispatch_delegate_task(
    +                parent,
    +                {
    +                    "goal": "test",
    +                    "acp_command": "claude",
    +                    "acp_args": ["--acp", "--stdio"],
    +                    "tasks": [
    +                        {
    +                            "goal": "nested",
    +                            "acp_command": "codex",
    +                            "acp_args": ["--acp"],
    +                        },
    +                    ],
    +                },
                 )
    -            _, kwargs = mock_build.call_args
    -            self.assertEqual(kwargs["override_acp_command"], "claude")
    -            self.assertEqual(kwargs["override_acp_args"], ["--acp", "--stdio"])
    +
    +        self.assertNotIn("acp_command", captured)
    +        self.assertNotIn("acp_args", captured)
    +        self.assertEqual(captured["goal"], "test")
    +        self.assertNotIn("acp_command", captured["tasks"][0])
    +        self.assertNotIn("acp_args", captured["tasks"][0])
     
     class TestDelegateEventEnum(unittest.TestCase):
         """Tests for DelegateEvent enum and back-compat aliases."""
    @@ -2600,31 +2551,15 @@ def test_schema_has_role_top_level_and_per_task(self):
             self.assertIn("role", task_props)
             self.assertEqual(task_props["role"]["enum"], ["leaf", "orchestrator"])
     
    -    def test_acp_command_description_has_do_not_set_guidance(self):
    -        # acp_command/acp_args descriptions must NOT bias the model toward
    -        # assuming an ACP CLI (Claude, Copilot, etc.) is installed. They must
    -        # carry explicit "do not set unless told" guidance so the model doesn't
    -        # hallucinate ACP availability (#22013).
    +    def test_schema_omits_acp_transport_fields(self):
             from tools.delegate_tool import DELEGATE_TASK_SCHEMA
             props = DELEGATE_TASK_SCHEMA["parameters"]["properties"]
     
    -        top_acp_desc = props["acp_command"]["description"]
    -        self.assertIn("Do NOT set", top_acp_desc)
    -        self.assertIn("explicitly told you", top_acp_desc)
    -
             task_props = props["tasks"]["items"]["properties"]
    -        per_task_acp_desc = task_props["acp_command"]["description"]
    -        self.assertIn("Do NOT set", per_task_acp_desc)
    -
    -    def test_acp_command_description_has_no_claude_as_example(self):
    -        # Descriptions must not list 'claude' as a canonical example value —
    -        # that directly primes the model to attempt Claude ACP even when it is
    -        # not installed (#22013).
    -        from tools.delegate_tool import DELEGATE_TASK_SCHEMA
    -        props = DELEGATE_TASK_SCHEMA["parameters"]["properties"]
    -        top_acp_desc = props["acp_command"]["description"].lower()
    -        self.assertNotIn("e.g. 'claude'", top_acp_desc)
    -        self.assertNotIn("e.g. \"claude\"", top_acp_desc)
    +        self.assertNotIn("acp_command", props)
    +        self.assertNotIn("acp_args", props)
    +        self.assertNotIn("acp_command", task_props)
    +        self.assertNotIn("acp_args", task_props)
     
     
     # Sentinel used to distinguish "role kwarg omitted" from "role=None".

  • tools/browser_camofox.py

    @@ -653,6 +653,10 @@ def camofox_click(ref: str, task_id: Optional[str] = None) -> str:
             if not session["tab_id"]:
                 return tool_error("No browser session. Call browser_navigate first.", success=False)
     
    +        blocked = _camofox_private_page_block(session, task_id, "click")
    +        if blocked:
    +            return blocked
    +
             # Strip @ prefix if present (our tool convention)
             clean_ref = ref.lstrip("@")
     
    @@ -676,6 +680,10 @@ def camofox_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
             if not session["tab_id"]:
                 return tool_error("No browser session. Call browser_navigate first.", success=False)
     
    +        blocked = _camofox_private_page_block(session, task_id, "type")
    +        if blocked:
    +            return blocked
    +
             clean_ref = ref.lstrip("@")
     
             _post(
    @@ -745,6 +753,10 @@ def camofox_press(key: str, task_id: Optional[str] = None) -> str:
             if not session["tab_id"]:
                 return tool_error("No browser session. Call browser_navigate first.", success=False)
     
    +        blocked = _camofox_private_page_block(session, task_id, "press")
    +        if blocked:
    +            return blocked
    +
             _post(
                 f"/tabs/{session['tab_id']}/press",
                 {"userId": session["user_id"], "key": key},

  • tools/browser_cdp_tool.py

    @@ -428,6 +428,15 @@ def browser_cdp(
     
         # --- Route iframe-scoped calls through the supervisor ---------------
         if frame_id:
    +        # Same private-page/SSRF boundary as the stateless path below —
    +        # frame_id routing must not become the sibling bypass for it.
    +        blocked = _browser_cdp_private_guard(
    +            task_id=effective_task_id,
    +            method=method,
    +            params=params or {},
    +        )
    +        if blocked:
    +            return blocked
             return _browser_cdp_via_supervisor(
                 task_id=effective_task_id,
                 frame_id=frame_id,

  • tools/delegate_tool.py

    @@ -1055,7 +1055,7 @@ def _build_child_agent(
         override_base_url: Optional[str] = None,
         override_api_key: [redacted:secret-assignment] = None,
         override_api_mode: Optional[str] = None,
    -    # ACP transport overrides — lets a non-ACP parent spawn ACP child agents
    +    # ACP transport overrides from trusted delegation config.
         override_acp_command: Optional[str] = None,
         override_acp_args: Optional[List[str]] = None,
         # Per-call role controlling whether the child can further delegate.
    @@ -1212,11 +1212,9 @@ def _child_thinking(text: str) -> None:
             effective_api_mode = None  # force re-derivation from provider's defaults
         else:
             effective_api_mode = getattr(parent_agent, "api_mode", None)
    -    # Defensive: validate override_acp_command exists on PATH before honoring
    -    # it. Models occasionally pass acp_command="copilot" / "claude" / etc. in
    -    # delegate_task tool calls despite the schema saying not to, which forces
    -    # the subagent onto the copilot-acp transport below and crashes the
    -    # gateway when the binary is missing (e.g. headless container deploys).
    +    # Defensive: validate trusted delegation.command exists on PATH before
    +    # honoring it. Stale config should not force a child onto the ACP transport
    +    # and then fail at subprocess startup.
         if override_acp_command:
             import shutil as _shutil
     
    @@ -2346,8 +2344,6 @@ def delegate_task(
         context: Optional[str] = None,
         tasks: Optional[List[Dict[str, Any]]] = None,
         max_iterations: Optional[int] = None,
    -    acp_command: Optional[str] = None,
    -    acp_args: Optional[List[str]] = None,
         role: Optional[str] = None,
         background: Optional[bool] = None,
         parent_agent=None,
    @@ -2486,7 +2482,6 @@ def delegate_task(
         children = []
         try:
             for i, t in enumerate(task_list):
    -            task_acp_args = t.get("acp_args") if "acp_args" in t else None
                 # Per-task role beats top-level; normalise again so unknown
                 # per-task values warn and degrade to leaf uniformly.
                 effective_role = _normalize_role(t.get("role") or top_role)
    @@ -2505,14 +2500,8 @@ def delegate_task(
                     override_base_url=creds["base_url"],
                     override_api_key=[redacted:secret-assignment]"api_key"],
                     override_api_mode=creds["api_mode"],
    -                override_acp_command=t.get("acp_command")
    -                or acp_command
    -                or creds.get("command"),
    -                override_acp_args=(
    -                    task_acp_args
    -                    if task_acp_args is not None
    -                    else (acp_args if acp_args is not None else creds.get("args"))
    -                ),
    +                override_acp_command=creds.get("command"),
    +                override_acp_args=creds.get("args"),
                     role=effective_role,
                 )
                 # Override with correct parent tool names (before child construction mutated global)
    @@ -3292,30 +3281,6 @@ def _build_role_param_description() -> str:
         )
     
     
    -# Known ACP-compatible CLIs that delegate_task can shell out to. Kept
    -# narrow on purpose: only the ones agent/copilot_acp_client.py and friends
    -# actually understand. Add new entries here when a new ACP CLI ships.
    -_KNOWN_ACP_BINARIES: tuple[str, ...] = ("copilot", "claude", "codex")
    -
    -
    -def _acp_binary_available() -> bool:
    -    """True iff at least one known ACP CLI is on PATH.
    -
    -    Used to gate inclusion of ``acp_command`` / ``acp_args`` in the
    -    delegate_task schema. On headless hosts (Railway / Fly / Docker /
    -    fresh VPS) without any of these binaries, exposing the fields invites
    -    the model to hallucinate ``acp_command="copilot"`` from the schema's
    -    description, which used to crash subagent runs and take the gateway
    -    down. Pruning the fields from the schema removes the temptation.
    -
    -    Not cached: ``shutil.which`` is cheap and we want the schema to react
    -    to mid-session installs without forcing a process restart.
    -    """
    -    import shutil as _shutil
    -
    -    return any(_shutil.which(name) for name in _KNOWN_ACP_BINARIES)
    -
    -
     def _build_dynamic_schema_overrides() -> dict:
         """Return per-call schema overrides reflecting current config.
     
    @@ -3333,24 +3298,6 @@ def _build_dynamic_schema_overrides() -> dict:
         overrides_params["properties"]["tasks"]["description"] = _build_tasks_param_description()
         overrides_params["properties"]["role"]["description"] = _build_role_param_description()
     
    -    # Prune ACP overrides from the schema when no known ACP CLI is on PATH.
    -    # The runtime guard in _build_child_agent remains as defense-in-depth for
    -    # internal callers / tests / future code paths that skip the schema layer.
    -    if not _acp_binary_available():
    -        overrides_params["properties"].pop("acp_command", None)
    -        overrides_params["properties"].pop("acp_args", None)
    -        tasks_schema = dict(overrides_params["properties"].get("tasks", {}))
    -        if "items" in tasks_schema:
    -            items = dict(tasks_schema["items"])
    -            if "properties" in items:
    -                items["properties"] = {
    -                    k: v
    -                    for k, v in items["properties"].items()
    -                    if k not in ("acp_command", "acp_args")
    -                }
    -            tasks_schema["items"] = items
    -            overrides_params["properties"]["tasks"] = tasks_schema
    -
         return {
             "description": _build_top_level_description(),
             "parameters": overrides_params,
    @@ -3401,19 +3348,6 @@ def _build_dynamic_schema_overrides() -> dict:
                                 "type": "string",
                                 "description": "Task-specific context",
                             },
    -                        "acp_command": {
    -                            "type": "string",
    -                            "description": (
    -                                "Per-task ACP command override (e.g. 'copilot'). "
    -                                "Overrides the top-level acp_command for this task only. "
    -                                "Do NOT set unless the user explicitly told you an ACP CLI is installed."
    -                            ),
    -                        },
    -                        "acp_args": {
    -                            "type": "array",
    -                            "items": {"type": "string"},
    -                            "description": "Per-task ACP args override. Leave empty unless acp_command is set.",
    -                        },
                             "role": {
                                 "type": "string",
                                 "enum": ["leaf", "orchestrator"],
    @@ -3444,28 +3378,6 @@ def _build_dynamic_schema_overrides() -> dict:
                         "compatibility."
                     ),
                 },
    -            "acp_command": {
    -                "type": "string",
    -                "description": (
    -                    "Override ACP command for child agents (e.g. 'copilot'). "
    -                    "When set, children use ACP subprocess transport instead of inheriting "
    -                    "the parent's transport. Requires an ACP-compatible CLI "
    -                    "(currently GitHub Copilot CLI via 'copilot --acp --stdio'). "
    -                    "See agent/copilot_acp_client.py for the implementation. "
    -                    "IMPORTANT: Do NOT set this unless the user has explicitly told you "
    -                    "a specific ACP-compatible CLI is installed and configured. "
    -                    "Leave empty to use the parent's default transport (Hermes subagents)."
    -                ),
    -            },
    -            "acp_args": {
    -                "type": "array",
    -                "items": {"type": "string"},
    -                "description": (
    -                    "Arguments for the ACP command (default: ['--acp', '--stdio']). "
    -                    "Only used when acp_command is set. "
    -                    "Leave empty unless acp_command is explicitly provided."
    -                ),
    -            },
             },
             "required": [],
         },
    @@ -3492,6 +3404,28 @@ def _model_background_value(args: dict, parent_agent=None) -> bool:
         return not is_subagent
     
     
    +_MODEL_HIDDEN_TASK_FIELDS = {"acp_command", "acp_args"}
    +
    +
    +def _strip_model_hidden_task_fields(tasks: Any) -> Any:
    +    if not isinstance(tasks, list):
    +        return tasks
    +    stripped_tasks = []
    +    changed = False
    +    for task in tasks:
    +        if not isinstance(task, dict):
    +            stripped_tasks.append(task)
    +            continue
    +        stripped = {
    +            key: value
    +            for key, value in task.items()
    +            if key not in _MODEL_HIDDEN_TASK_FIELDS
    +        }
    +        changed = changed or len(stripped) != len(task)
    +        stripped_tasks.append(stripped)
    +    return stripped_tasks if changed else tasks
    +
    +
     registry.register(
         name="delegate_task",
         toolset="delegation",
    @@ -3499,10 +3433,8 @@ def _model_background_value(args: dict, parent_agent=None) -> bool:
         handler=lambda args, **kw: delegate_task(
             goal=args.get("goal"),
             context=args.get("context"),
    -        tasks=args.get("tasks"),
    +        tasks=_strip_model_hidden_task_fields(args.get("tasks")),
             max_iterations=args.get("max_iterations"),
    -        acp_command=args.get("acp_command"),
    -        acp_args=args.get("acp_args"),
             role=args.get("role"),
             background=_model_background_value(args, kw.get("parent_agent")),
             parent_agent=kw.get("parent_agent"),