gitea-admin/cowork-local
2
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(R05): tool capability registry, unified policy gateway, MCP lifecycle manager

Browse Source 
EPIC R05 (Team Hoa) - one security/approval path for every tool call.

R05-T01 domain/tools/{tool_descriptor,tool_registry}.py
  ToolCapability (READ/WRITE/EXECUTE/NETWORK, composable) + ToolDescriptor +
  ToolRegistry, replacing three independently-maintained gating lists
  (core/tools.py::WRITE_TOOLS, code_agent.py's WRITE_TOOLS|MS365_WRITE_TOOLS,
  chat_agent.py's literal ("run_command","install_package") tuple) with one
  capability lookup.

R05-T02 infrastructure/filesystem/{file_tools,command_tools,fetch_tools,tool_context}.py
  core/tools.py's execute_tool if/elif chain split into per-concern modules.
  core/tools.py is now a strangler-fig shim: re-exports ToolContext/ToolError,
  dispatches through a {name: handler} dict built from the split modules.
  core/tools.py: 566 -> 291 lines.

R05-T03 application/conversations/tool_policy_gateway.py
  ToolPolicyGateway.allow(name, gate, payload) - capability-driven ALLOW vs
  ask-the-gate decision. Wired into both chat_agent.py::run_cowork and
  code_agent.py::run_code, replacing their separate hand-rolled checks.
  Verified equivalent to the old hardcoded sets by test.

R05-T04 (behavior change, not just refactor)
  MCP/connector tools (core/mcp_client.py, core/ext_connectors.py) reached
  chat_agent.py via extra_executor(name, args) with NO permission check at
  all. They are now tagged with a conservative default capability
  (WRITE|EXECUTE|NETWORK - no MCP tool self-declares risk) and routed through
  the SAME ToolPolicyGateway as built-ins. When "confirm before running
  commands" is on, MCP/connector calls now prompt like run_command already
  did - a real gap closed, and a user-visible change worth calling out.

R05-T05 infrastructure/mcp/mcp_source_manager.py
  McpToolSourceManager extracts the connection cache/lock/start-or-skip
  lifecycle out of state.py::AppContext (_mcp_connections/_conn_lock) into a
  standalone, directly-testable class. AppContext.build_mcp_tools and
  _ms365_builtin_connection now call ensure()/stop(); _ext_connections
  (unified Connectors) is out of scope for this task and keeps its own lock.

New tests: tests/unit/test_tool_registry_and_policy.py,
test_code_agent_tool_policy.py, test_cowork_extra_tool_policy.py,
test_mcp_source_manager.py (26 new tests).

Suite: 254 passed, 4 pre-existing failures unrelated to R05 (2 EPIC R02
config-security, 2 environment-dependent routing tests - see checklist).
check_imports: PASS. All new files < 400 LOC.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
vudt15
2026-08-21 22:20:57 +09:00
co-authored by Claude Sonnet 5
parent 6d3217e0b5
commit ae4fe72b2e
22 changed files with 1355 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/chat_agent.py
+46 -10
View File
@@ -11,6 +11,8 @@ import re
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from ..application.conversations.tool_policy_gateway import ToolPolicyGateway
from ..domain.tools import ToolCapability, default_registry
from ..providers.base import Provider, ToolSpec
from . import agent_roles
from . import agent_security
@@ -27,6 +29,13 @@ from .tools import TOOL_SPECS, ToolContext, _snapshot, describe_action, execute_
# Generator / helper scripts — never a final deliverable in Cowork's output.
_SCRIPT_EXTS = {".py", ".pyw", ".js", ".mjs", ".cjs", ".ts", ".sh", ".bat", ".ps1", ".rb", ".pl"}
# R05-T03/T04: replaces the literal ``name in ("run_command",
# "install_package")`` check below with a capability lookup — EXECUTE is
# exactly the capability those two (and only those two) built-in tools carry
# (see domain/tools/tool_registry.py::BUILT_IN_CAPABILITIES). Copied per-turn
# into ``turn_tool_policy`` inside run_cowork() once extra_tools are known.
_COWORK_TOOL_REGISTRY = default_registry(TOOL_SPECS)
EmitFn = Callable[[Dict[str, Any]], None]
CancelFn = Callable[[], bool]
@@ -388,6 +397,19 @@ def run_cowork(
jira=(security_config.data.get("jira") if security_config else None))
extra_tools = extra_tools or []
extra_names = {t.name for t in extra_tools}
# R05-T04: MCP servers (core/mcp_client.py) and unified connectors
# (core/ext_connectors.py) — everything that arrives here as extra_tools —
# advertise no standard risk metadata, so each is tagged with the same
# conservative default (WRITE|EXECUTE|NETWORK) domain/tools/tool_registry.py
# uses for any unclassified tool. Copying the built-in registry per turn
# (cheap - under 20 entries) rather than mutating the shared module-level
# one keeps different turns' extra_tools from leaking into each other.
from ..domain.tools import ToolDescriptor, ToolRegistry
from ..domain.tools.tool_registry import UNKNOWN_SOURCE_CAPABILITIES
_turn_registry = ToolRegistry(_COWORK_TOOL_REGISTRY.all())
for _spec in extra_tools:
_turn_registry.register(ToolDescriptor.from_spec(_spec, UNKNOWN_SOURCE_CAPABILITIES))
turn_tool_policy = ToolPolicyGateway(_turn_registry, ToolCapability.EXECUTE)
# update_plan drives the Plan panel (above Output); it produces no file.
# Built-in tools the admin disabled (Monitoring → Tools) are filtered out.
from .tools import enabled_tool_specs
@@ -489,6 +511,18 @@ def run_cowork(
preview = {"kind": "info", "title": name, "text": str(args)}
emit({"type": "tool_proposed", "id": tc_id, "name": name, "args": args,
"preview": preview})
# R05-T04: MCP/connector tools used to run with NO permission
# check at all — this is what closes that gap. Same policy,
# same gate object as the built-in tools below.
if not turn_tool_policy.allow(
name, gate, {"name": name, "args": args, "preview": preview}
):
result = {"ok": False, "output": "Rejected by user."}
emit({"type": "tool_result", "id": tc_id, "name": name,
"ok": False, "output": result["output"]})
messages.append({"role": "tool", "tool_call_id": tc_id, "name": name,
"content": result["output"]})
continue
result = extra_executor(name, args)
emit({"type": "tool_result", "id": tc_id, "name": name,
"ok": result.get("ok", False), "output": result.get("output", "")})
@@ -528,16 +562,18 @@ def run_cowork(
# Permission Management (Sandbox Security Layer) — only when a
# gate was actually supplied (Settings: "confirm before running
# commands"); None preserves the pre-existing auto-run behavior.
if gate is not None and name in ("run_command", "install_package"):
approved = gate.request({"name": name, "args": args, "preview": preview})
if not approved:
result = {"ok": False, "output": "Rejected by user."}
evt = {"type": "tool_result", "id": tc_id, "name": name,
"ok": False, "output": result["output"]}
emit(evt)
messages.append({"role": "tool", "tool_call_id": tc_id,
"name": name, "content": result["output"]})
continue
# R05-T03: gating is now capability-driven (see
# turn_tool_policy above) instead of a literal name tuple.
if not turn_tool_policy.allow(
name, gate, {"name": name, "args": args, "preview": preview}
):
result = {"ok": False, "output": "Rejected by user."}
evt = {"type": "tool_result", "id": tc_id, "name": name,
"ok": False, "output": result["output"]}
emit(evt)
messages.append({"role": "tool", "tool_call_id": tc_id,
"name": name, "content": result["output"]})
continue
if name == "save_file":
result = _do_save_file(output_dir, title, args)
core/code_agent.py
+15 -4
View File
@@ -12,6 +12,8 @@ import re
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from ..application.conversations.tool_policy_gateway import ToolPolicyGateway
from ..domain.tools import ToolCapability, ToolDescriptor, ToolRegistry
from ..providers.base import Provider
from . import agent_roles
from . import agent_security
@@ -225,6 +227,14 @@ def run_code(
# read/list ms365 tools count as "read-only, never confirm". Names are
# the MCP-qualified "ms365__*" form the agent sees (see ms365_tools.py).
gated_tools = WRITE_TOOLS | MS365_WRITE_TOOLS
# R05-T03/T04: ``gated_tools`` stays the authoritative name set (unchanged),
# but the actual confirm decision now goes through the same
# ToolPolicyGateway class run_cowork uses, instead of a separate
# hand-rolled ``if name in gated_tools`` + direct ``gate.request(...)``.
code_tool_policy = ToolPolicyGateway(
ToolRegistry(ToolDescriptor(n, "", {}, ToolCapability.WRITE) for n in gated_tools),
ToolCapability.WRITE,
)
# In PLAN mode, don't advertise write/run tools (analysis only).
advertised = [t for t in all_tools if t.name not in gated_tools] if plan else all_tools
has_memory = any(t.name.startswith("cmem_") for t in extra_tools)
@@ -297,10 +307,11 @@ def run_code(
agent_security.enforce_command(provider, name, args, security_config, emit,
agent_kind="code")
if name in gated_tools:
approved = gate.request({"id": tc_id, "name": name, "args": args, "preview": preview})
else:
approved = True # read-only tools (incl. codebase memory) never confirm
# read-only tools (incl. codebase memory) never consult the gate —
# code_tool_policy.requires_confirmation(name) is False for them.
approved = code_tool_policy.allow(
name, gate, {"id": tc_id, "name": name, "args": args, "preview": preview}
)
if cancel():
return messages
core/mcp_client.py
+7
View File
@@ -106,6 +106,13 @@ class McpServerConnection:
if self._thread is not None:
self._thread.join(timeout=5)
def is_alive(self) -> bool:
"""True while the connection's background thread (and therefore its
event loop and subprocess) is still running — used by
``infrastructure/mcp/mcp_source_manager.py`` (R05-T05) to tell a live
cached connection from one whose subprocess already died."""
return self._thread is not None and self._thread.is_alive()
# ---- tools -----------------------------------------------------------
def list_tool_specs(self) -> List[ToolSpec]:
"""The server's tools, wrapped as :class:`ToolSpec` — the same shape
core/tools.py
+38 -312
View File
@@ -3,79 +3,29 @@
Every path is resolved relative to the working directory and must stay inside
it (path-traversal is rejected). ``run_command`` executes inside the workdir
with a timeout and captured output.
R05-T02: the actual handlers (``read_file``/``list_dir``/``write_file``/
``edit_file``/``run_command``/``install_package``/``fetch_url``/
``jira_search``/``jira_get_issue``) now live in
``infrastructure/filesystem/{file_tools,command_tools,fetch_tools}.py``, split
out of what used to be one big if/elif chain here. This module is the
strangler-fig shim (ADR-001 section 4): it re-exports ``ToolContext``/
``ToolError`` (actually defined in
``infrastructure/filesystem/tool_context.py`` now) so every existing
``from .tools import ToolContext`` keeps working, and ``execute_tool``
dispatches through a small ``{name: handler}`` table built from the moved
modules instead of the chain itself.
"""
from __future__ import annotations
import ast
import difflib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from ..infrastructure.filesystem import command_tools, fetch_tools, file_tools
from ..infrastructure.filesystem.command_tools import _snapshot # noqa: F401 - re-export, core/chat_agent.py imports this name
from ..infrastructure.filesystem.tool_context import CancelFn, ToolContext, ToolError # noqa: F401 - re-export
from ..providers.base import ToolSpec
CancelFn = Callable[[], bool]
MAX_READ_BYTES = 200_000
COMMAND_TIMEOUT = 120 # seconds
class ToolError(Exception):
pass
def _flatten_rel(rel: str) -> str:
"""Collapse a sub-folder path down to a bare filename so the file lands in the
workdir root — EXCEPT the ``.scratch`` sandbox subtree, which is preserved.
Used by the Cowork agent (flatten_writes=True) so it can never create a
per-session / per-chat / per-task output sub-folder: every deliverable stays
directly in the single configured Output folder."""
parts = Path(rel).parts
if parts and parts[0] == ".scratch":
return rel # temporary sandbox is allowed (and cleaned up afterwards)
return Path(rel).name or rel
@dataclass
class ToolContext:
workdir: Path
flatten_writes: bool = False # Cowork: force every write into the workdir root
sandbox: bool = False # Code tab: isolate run_command/install_package into <workdir>/.venv
# Sandbox Security Layer — Settings' "Resource Limits" (cpu_percent/memory_mb/
# disk_mb), applied to every run_command/install_package this context runs.
# None (default) = no limits, matching pre-existing behavior.
resource_limits: Optional[Dict[str, float]] = None
# Sandbox Security Layer — Settings' "Block network for agent commands"
# (policy-level, see deps.py::network_blocked_env). False (default) =
# unrestricted, matching pre-existing behavior.
block_network: bool = False
# Whether the fetch_url tool may read URLs — SEPARATE from block_network
# (reading a web page/share link for info is safe; running networked shell
# commands is the risk). Defaults True; set from agent_security.allow_url_fetch.
allow_url_fetch: bool = True
# Jira read connector config (base_url/email/api_token) — None disables the
# jira_* tools' ability to connect. Populated from config.data["jira"].
jira: Optional[Dict[str, Any]] = None
def resolve(self, rel: str) -> Path:
"""Resolve ``rel`` inside the workdir, rejecting escapes."""
if rel in ("", "."):
return self.workdir
candidate = (self.workdir / rel).expanduser()
try:
resolved = candidate.resolve()
except OSError as exc:
raise ToolError(f"Invalid path: {rel} ({exc})")
root = self.workdir.resolve()
if resolved != root and root not in resolved.parents:
raise ToolError(
f"Refused: '{rel}' is outside the working folder ({root})."
)
return resolved
# --------------------------------------------------------------------------
# Tool specs advertised to the model
# --------------------------------------------------------------------------
@@ -192,6 +142,23 @@ TOOL_SPECS: List[ToolSpec] = [
# Actions gated by the permission gate in confirm mode (auto-approved in Auto-run).
WRITE_TOOLS = {"write_file", "edit_file", "run_command", "install_package"}
# name -> handler(ctx, args[, cancel, on_output]) — built once from the split
# infrastructure modules. Replaces the if/elif chain execute_tool used to be.
_HANDLERS: Dict[str, Callable[..., Dict[str, Any]]] = {
"read_file": file_tools.read_file,
"list_dir": file_tools.list_dir,
"write_file": file_tools.write_file,
"edit_file": file_tools.edit_file,
"run_command": command_tools.run_command,
"install_package": command_tools.install_package,
"fetch_url": fetch_tools.fetch_url,
"jira_search": fetch_tools.jira_search,
"jira_get_issue": fetch_tools.jira_get_issue,
}
# Handlers that accept the long-running (cancel, on_output) signature — every
# other handler takes just (ctx, args).
_CANCELLABLE = {"run_command", "install_package"}
def enabled_tool_specs(security_config=None) -> List[ToolSpec]:
"""The built-in TOOL_SPECS minus any the admin turned OFF in Monitoring →
@@ -301,27 +268,14 @@ def execute_tool(ctx: ToolContext, name: str, args: Dict[str, Any],
labels WHICH agent role made it."""
from . import audit_log
handler = _HANDLERS.get(name)
try:
if name == "read_file":
result = _read_file(ctx, args)
elif name == "list_dir":
result = _list_dir(ctx, args)
elif name == "write_file":
result = _write_file(ctx, args)
elif name == "edit_file":
result = _edit_file(ctx, args)
elif name == "run_command":
result = _run_command(ctx, args, cancel, on_output)
elif name == "install_package":
result = _install_package(ctx, args, cancel, on_output)
elif name == "fetch_url":
result = _fetch_url(ctx, args)
elif name == "jira_search":
result = _jira_search(ctx, args)
elif name == "jira_get_issue":
result = _jira_get_issue(ctx, args)
else:
if handler is None:
result = {"ok": False, "output": f"Tool not found: {name}"}
elif name in _CANCELLABLE:
result = handler(ctx, args, cancel, on_output)
else:
result = handler(ctx, args)
except ToolError as exc:
result = {"ok": False, "output": str(exc)}
except Exception as exc: # defensive: a tool must never crash the agent
@@ -331,234 +285,6 @@ def execute_tool(ctx: ToolContext, name: str, args: Dict[str, Any],
return result
def _fetch_url(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Fetch a URL's text content (web page / online document / SharePoint-
OneDrive share link) via link_fetch — the same parser task-link attachments
use. Honors the Sandbox Security Layer's "Block network" policy."""
url = str(args.get("url", "")).strip()
if not url:
return {"ok": False, "output": "fetch_url: 'url' is required."}
if not url.lower().startswith(("http://", "https://")):
return {"ok": False, "output": f"fetch_url: not an http(s) URL: {url}"}
if not ctx.allow_url_fetch:
return {"ok": False,
"output": ("fetch_url: URL fetching is turned off in Settings → Security "
"(\"Allow the agent to fetch URLs\").")}
# A pasted Jira issue link on the CONNECTED Jira host is read via the
# authenticated API (so private issues resolve, not a login page). Public
# links / any other URL fall through to the normal fetcher below.
from . import jira_tool
if jira_tool.is_jira_issue_url(ctx.jira, url):
return {"ok": True, "output": jira_tool.get_issue_by_url(ctx.jira, url)}
from .link_fetch import fetch_link_preview
return {"ok": True, "output": fetch_link_preview(url)}
def _jira_search(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from . import jira_tool
out = jira_tool.search(ctx.jira, str(args.get("jql", "")),
int(args.get("max_results", 25) or 25))
return {"ok": not out.lower().startswith(("jira is not configured", "jira search failed")),
"output": out}
def _jira_get_issue(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
from . import jira_tool
out = jira_tool.get_issue(ctx.jira, str(args.get("key", "")))
return {"ok": not out.lower().startswith(("jira is not configured", "could not fetch")),
"output": out}
def _read_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
target = ctx.resolve(str(args.get("path", "")))
if not target.exists():
return {"ok": False, "output": f"File not found: {args.get('path')}"}
data = target.read_bytes()[:MAX_READ_BYTES]
text = data.decode("utf-8", errors="replace")
return {"ok": True, "output": text}
def _list_dir(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ".") or ".")
target = ctx.resolve(rel)
# A missing/not-yet-created path is NOT a tool failure — report it as an
# ordinary result so the agent can create it or pick another path and keep
# going. Returning ok=False here surfaced a false "tool failed: list_dir" in
# Co4E flows and could stall a step on a recoverable situation.
if not target.exists():
return {"ok": True, "output": f"(path '{rel}' does not exist yet — create it or use another path)"}
if target.is_file():
return {"ok": True, "output": f"('{rel}' is a file, not a directory)"}
entries = []
for child in sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
marker = "/" if child.is_dir() else ""
entries.append(f"{child.name}{marker}")
return {"ok": True, "output": "\n".join(entries) or "(empty folder)"}
def _check_python_syntax(target: Path, content: str) -> str:
"""Return a short warning if ``content`` is invalid Python, else ''.
Catches syntax errors the instant a .py file is written/edited — before the
agent wastes a whole run_command round-trip just to get the same error back
from a traceback."""
if target.suffix.lower() not in (".py", ".pyw"):
return ""
try:
ast.parse(content, filename=str(target))
return ""
except SyntaxError as exc:
return f"\n⚠ Syntax error at line {exc.lineno}: {exc.msg} — fix this before running the file."
def _write_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
content = str(args.get("content", ""))
target.parent.mkdir(parents=True, exist_ok=True)
# A .xlsx is a binary package — build a REAL workbook from the content
# (CSV/TSV/Markdown-table/JSON) rather than writing raw text (which corrupts it).
if target.suffix.lower() in (".xlsx", ".xlsm"):
from . import xlsx_write
if xlsx_write.build_xlsx_from_text(target, content):
return {"ok": True, "path": str(target),
"output": f"Wrote spreadsheet {rel} ({target.name})."}
return {"ok": False, "output": "Could not build the .xlsx (openpyxl unavailable) — "
"write a .csv instead, or use a generator script."}
target.write_text(content, encoding="utf-8")
warning = _check_python_syntax(target, content)
return {"ok": True, "path": str(target),
"output": f"Wrote {len(content)} chars to {rel}.{warning}"}
def _edit_file(ctx: ToolContext, args: Dict[str, Any]) -> Dict[str, Any]:
"""Replace an exact snippet inside an existing file (precise patch edit)."""
rel = str(args.get("path", ""))
if ctx.flatten_writes:
rel = _flatten_rel(rel)
target = ctx.resolve(rel)
if not target.exists():
return {"ok": False,
"output": f"File not found: {rel} — use write_file to create it."}
old = str(args.get("old_string", ""))
new = str(args.get("new_string", ""))
replace_all = bool(args.get("replace_all", False))
if not old:
return {"ok": False, "output": "old_string is empty — provide the exact text to replace."}
try:
text = target.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
return {"ok": False, "output": f"Could not read file: {exc}"}
count = text.count(old)
if count == 0:
return {"ok": False, "output": ("old_string not found. Read the file and copy the exact "
"text to replace, including indentation/whitespace.")}
if count > 1 and not replace_all:
return {"ok": False, "output": (f"old_string appears {count} times — add surrounding "
"context to make it unique, or set replace_all=true.")}
updated = text.replace(old, new) if replace_all else text.replace(old, new, 1)
target.write_text(updated, encoding="utf-8")
n = count if replace_all else 1
warning = _check_python_syntax(target, updated)
return {"ok": True,
"output": f"Edited {args.get('path')} ({n} replacement{'' if n == 1 else 's'}).{warning}"}
def _sandbox_python(ctx: ToolContext, cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Optional[str]:
"""Lazily create/reuse this ctx's project sandbox venv (Code tab only —
``ctx.sandbox``); returns its python path, or None to use the app's own."""
if not ctx.sandbox:
return None
from .deps import ensure_project_venv
py = ensure_project_venv(ctx.workdir, cancel=cancel, on_output=on_output)
return str(py) if py else None
def _install_package(ctx: ToolContext, args: Dict[str, Any], cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
from .deps import pip_install
package = str(args.get("package", "")).strip()
if not package:
return {"ok": False, "output": "No package specified."}
python = _sandbox_python(ctx, cancel, on_output)
ok, detail = pip_install(package, cancel=cancel, on_output=on_output, python=python)
head = f"Installed {package}." if ok else f"Could not install {package}."
return {"ok": ok, "output": f"{head}\n{detail}"}
_SNAPSHOT_SKIP = {".git", "__pycache__", "node_modules", ".scratch", ".venv",
".idea", ".mypy_cache", ".pytest_cache"}
def _snapshot(workdir: Path) -> Dict[str, Any]:
"""Map of file path -> (mtime, size) under the workdir (noise dirs skipped)."""
snap: Dict[str, Any] = {}
try:
for dirpath, dirnames, filenames in os.walk(str(workdir)):
dirnames[:] = [d for d in dirnames if d not in _SNAPSHOT_SKIP]
for fn in filenames:
full = os.path.join(dirpath, fn)
try:
st = os.stat(full)
snap[full] = (st.st_mtime_ns, st.st_size)
except OSError:
pass
if len(snap) > 5000:
return snap
except OSError:
pass
return snap
def _run_command(ctx: ToolContext, args: Dict[str, Any],
cancel: Optional[CancelFn] = None,
on_output: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
from .deps import network_blocked_env, run_cancellable, sandbox_env
from .sandbox_manager import SandboxManager, ExecutionConfig
from ..security.command_risk_classifier import classify_command
command = str(args.get("command", "")).strip()
if not command:
return {"ok": False, "output": "Empty command."}
# --- Security validation pipeline ---
risk = classify_command(command, is_cowork_mode=ctx.flatten_writes)
if risk.blocked:
denial = "Command blocked by security policy: " + "; ".join(risk.reasons)
return {"ok": False, "output": denial}
# Route through SandboxManager for risk-based isolation
mgr = SandboxManager(ExecutionConfig(
enabled=True,
block_network_by_default=ctx.block_network,
is_cowork_mode=ctx.flatten_writes,
))
sandbox_result = mgr.run(
command=command,
workdir=str(ctx.workdir),
block_network=ctx.block_network,
timeout_sec=COMMAND_TIMEOUT,
cancel=cancel,
)
# Sandbox ALWAYS executes (never double-run). Return its result directly.
if sandbox_result.get("sandbox") == "blocked":
return {"ok": False, "output": sandbox_result.get("stderr", "Command blocked")}
out = sandbox_result.get("stdout", "").strip() or "(no output)"
err = sandbox_result.get("stderr", "")
rc = sandbox_result.get("returncode", -1)
if err:
out = f"{out}\n{err}" if out else err
return {"ok": sandbox_result.get("ok", False), "output": f"[exit {rc}]\n{out}"}
def _short_json(obj: Any, limit: int = 500) -> str:
import json
text = json.dumps(obj, ensure_ascii=False, indent=2)