fix(tools): report sandbox failures deterministically

This commit is contained in:
rookiestar28
2026-06-04 15:08:32 +08:00
parent 9784328e40
commit a46a7db84b
4 changed files with 220 additions and 24 deletions
+3 -1
View File
@@ -107,7 +107,9 @@ def resolve_package_resource_path(
relative_path = _PACKAGE_RESOURCES[resource_id]
except KeyError as exc:
known = ", ".join(sorted(_PACKAGE_RESOURCES))
raise KeyError(f"unknown package resource '{resource_id}'; known: {known}") from exc
raise KeyError(
f"unknown package resource '{resource_id}'; known: {known}"
) from exc
root = (
Path(package_root).resolve()
+96 -21
View File
@@ -24,6 +24,50 @@ DEFAULT_TOOL_MAX_OUTPUT_BYTES = 64 * 1024 # 64KB
_TRUTHY = {"1", "true", "yes", "on"}
_SANDBOX_RUNTIME_ENV = "OPENCLAW_TOOL_SANDBOX_RUNTIME_AVAILABLE"
TOOL_ERROR_TOOL_NOT_ALLOWED = "tool_not_allowed"
TOOL_ERROR_SANDBOX_RUNTIME_UNAVAILABLE = "sandbox_runtime_unavailable"
TOOL_ERROR_SANDBOX_POLICY_MISSING = "sandbox_policy_missing"
TOOL_ERROR_NETWORK_HOSTS_MISSING = "network_hosts_missing"
TOOL_ERROR_WORKSPACE_VIOLATION = "workspace_violation"
TOOL_ERROR_INTERPRETER_MISSING = "interpreter_missing"
TOOL_ERROR_TIMEOUT = "timeout"
TOOL_ERROR_PROCESS_FAILED = "process_failed"
TOOL_ERROR_EXECUTION_ERROR = "execution_error"
_TOOL_ERROR_REMEDIATIONS = {
TOOL_ERROR_TOOL_NOT_ALLOWED: (
"Add the tool to the allowlist or use an allowed tool name before retrying."
),
TOOL_ERROR_SANDBOX_RUNTIME_UNAVAILABLE: (
f"Set {_SANDBOX_RUNTIME_ENV}=1 only after the sandbox runtime is available, "
"or leave hardened mode only in a trusted local environment."
),
TOOL_ERROR_SANDBOX_POLICY_MISSING: (
"Define an explicit sandbox block for the tool in the allowlist."
),
TOOL_ERROR_NETWORK_HOSTS_MISSING: (
"Add allow_network_hosts for the network-enabled tool or disable network access."
),
TOOL_ERROR_WORKSPACE_VIOLATION: (
"Use paths inside the allowed filesystem prefixes or update the tool sandbox "
"allowlist."
),
TOOL_ERROR_INTERPRETER_MISSING: (
"Install the executable referenced by the tool allowlist or update the tool "
"command path."
),
TOOL_ERROR_TIMEOUT: (
"Review the tool process or increase timeout_sec only when the command is "
"expected to run longer."
),
TOOL_ERROR_PROCESS_FAILED: (
"Inspect the redacted tool output and allowlist configuration before retrying."
),
TOOL_ERROR_EXECUTION_ERROR: (
"Review the tool allowlist, validated arguments, and local runtime setup."
),
}
def is_tools_enabled() -> bool:
"""Check if external tooling is enabled (Opt-in)."""
@@ -35,6 +79,27 @@ def is_tools_enabled() -> bool:
)
def _diagnostic_failure_result(
*,
tool_name: str,
duration_ms: float,
error: str,
error_code: str,
output: str = "",
exit_code: Optional[int] = None,
) -> "ToolResult":
return ToolResult(
tool_name=tool_name,
success=False,
output=output,
duration_ms=duration_ms,
error=error,
exit_code=exit_code,
error_code=error_code,
remediation=_TOOL_ERROR_REMEDIATIONS[error_code],
)
@dataclass
class ToolResult:
"""Result of a tool execution."""
@@ -45,6 +110,8 @@ class ToolResult:
duration_ms: float
error: Optional[str] = None
exit_code: Optional[int] = None
error_code: Optional[str] = None
remediation: Optional[str] = None
@dataclass
@@ -341,48 +408,44 @@ class ToolRunner:
tool = self._tools.get(tool_name)
if not tool:
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=0,
error=f"Tool '{tool_name}' not allowed or not found.",
error_code=TOOL_ERROR_TOOL_NOT_ALLOWED,
)
hardened = self._is_hardened_mode()
if hardened:
# CRITICAL: hardened profile must fail closed when sandbox posture is ambiguous.
if not self._sandbox_runtime_available():
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=0,
error=(
"Sandbox runtime unavailable in hardened mode. "
f"Set {_SANDBOX_RUNTIME_ENV}=1."
),
error_code=TOOL_ERROR_SANDBOX_RUNTIME_UNAVAILABLE,
)
if not tool.sandbox_declared:
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=0,
error=(
"Missing explicit sandbox policy for tool in hardened mode. "
"Define a 'sandbox' block in tools allowlist."
),
error_code=TOOL_ERROR_SANDBOX_POLICY_MISSING,
)
if tool.sandbox.network and not tool.sandbox.allow_network_hosts:
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=0,
error=(
"Network-enabled tool requires allow_network_hosts in hardened mode."
),
error_code=TOOL_ERROR_NETWORK_HOSTS_MISSING,
)
try:
@@ -434,12 +497,11 @@ class ToolRunner:
tool_name,
violation_msg,
)
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=(time.monotonic() - start_time) * 1000,
error=f"Sandbox FS violation: {violation_msg}",
error_code=TOOL_ERROR_WORKSPACE_VIOLATION,
)
# Execute under sandbox workspace.
@@ -476,27 +538,40 @@ class ToolRunner:
error=(
None if success else f"Process exited with code {proc.returncode}"
),
error_code=None if success else TOOL_ERROR_PROCESS_FAILED,
remediation=(
None
if success
else _TOOL_ERROR_REMEDIATIONS[TOOL_ERROR_PROCESS_FAILED]
),
)
except subprocess.TimeoutExpired:
elapsed_ms = (time.monotonic() - start_time) * 1000
logger.warning(f"Tool '{tool_name}' timed out after {tool.timeout_sec}s")
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=elapsed_ms,
error="Execution timed out",
error_code=TOOL_ERROR_TIMEOUT,
)
except FileNotFoundError:
elapsed_ms = (time.monotonic() - start_time) * 1000
logger.error("Executable for tool '%s' was not found", tool_name)
return _diagnostic_failure_result(
tool_name=tool_name,
duration_ms=elapsed_ms,
error=f"Executable not found for tool '{tool_name}'.",
error_code=TOOL_ERROR_INTERPRETER_MISSING,
)
except Exception as e:
elapsed_ms = (time.monotonic() - start_time) * 1000
logger.error(f"Error executing tool '{tool_name}': {e}")
return ToolResult(
return _diagnostic_failure_result(
tool_name=tool_name,
success=False,
output="",
duration_ms=elapsed_ms,
error=str(e),
error_code=TOOL_ERROR_EXECUTION_ERROR,
)
@@ -6,7 +6,6 @@ from unittest.mock import patch
from services.tool_runner import ToolRunner
ROOT = Path(__file__).resolve().parents[1]
@@ -102,7 +101,9 @@ class TestR191RuntimeDependencyHygiene(unittest.TestCase):
runner = ToolRunner()
self.assertEqual(Path(runner._config_path), custom_allowlist)
self.assertEqual({"custom_echo"}, {tool["name"] for tool in runner.list_tools()})
self.assertEqual(
{"custom_echo"}, {tool["name"] for tool in runner.list_tools()}
)
def test_runtime_cache_contract_keeps_generated_paths_separate(self):
try:
+118
View File
@@ -0,0 +1,118 @@
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from services.tool_runner import SandboxProfile, ToolDefinition, ToolRunner
def _runner_with_tool(tool: ToolDefinition) -> ToolRunner:
runner = ToolRunner(config_path="dummy_path")
runner._tools[tool.name] = tool
return runner
def _echo_tool(name: str = "diagnostic_echo") -> ToolDefinition:
return ToolDefinition(
name=name,
command_template=["echo", "{msg}"],
allowed_args={"msg": "^[a-z]+$"},
timeout_sec=1,
max_output_bytes=100,
sandbox=SandboxProfile.strict(),
sandbox_declared=True,
)
class TestR192ToolSandboxDiagnostics(unittest.TestCase):
def test_hardened_missing_sandbox_runtime_is_coded_and_fail_closed(self):
runner = _runner_with_tool(_echo_tool())
with (
patch.dict(
"os.environ",
{
"OPENCLAW_RUNTIME_PROFILE": "hardened",
"OPENCLAW_TOOL_SANDBOX_RUNTIME_AVAILABLE": "0",
},
),
patch("subprocess.run") as mock_run,
):
result = runner.execute_tool("diagnostic_echo", {"msg": "hello"})
self.assertFalse(result.success)
self.assertEqual(
getattr(result, "error_code", None),
"sandbox_runtime_unavailable",
)
self.assertIn("OPENCLAW_TOOL_SANDBOX_RUNTIME_AVAILABLE", result.remediation)
mock_run.assert_not_called()
def test_missing_interpreter_is_coded_and_actionable(self):
runner = _runner_with_tool(_echo_tool())
with patch("subprocess.run") as mock_run:
mock_run.side_effect = FileNotFoundError("executable not found")
result = runner.execute_tool("diagnostic_echo", {"msg": "hello"})
self.assertFalse(result.success)
self.assertEqual(getattr(result, "error_code", None), "interpreter_missing")
self.assertIn("Install the executable", result.remediation)
def test_timeout_preserves_error_text_and_adds_diagnostics(self):
runner = _runner_with_tool(_echo_tool())
with patch("subprocess.run") as mock_run:
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["echo"], timeout=1)
result = runner.execute_tool("diagnostic_echo", {"msg": "hello"})
self.assertFalse(result.success)
self.assertEqual(result.error, "Execution timed out")
self.assertEqual(getattr(result, "error_code", None), "timeout")
self.assertIn("timeout", result.remediation.lower())
def test_workspace_violation_is_coded_and_does_not_execute(self):
with tempfile.TemporaryDirectory() as temp_dir:
allowed = Path(temp_dir) / "allowed"
denied = Path(temp_dir) / "denied" / "input.txt"
allowed.mkdir()
denied.parent.mkdir()
denied.write_text("blocked", encoding="utf-8")
tool = ToolDefinition(
name="path_reader",
command_template=["echo", "{path}"],
allowed_args={"path": r"^.+$"},
timeout_sec=1,
sandbox=SandboxProfile(allow_fs_read=[str(allowed)]),
sandbox_declared=True,
)
runner = _runner_with_tool(tool)
with patch("subprocess.run") as mock_run:
result = runner.execute_tool("path_reader", {"path": str(denied)})
self.assertFalse(result.success)
self.assertEqual(getattr(result, "error_code", None), "workspace_violation")
self.assertIn("allowed filesystem", result.remediation)
mock_run.assert_not_called()
def test_allowed_simple_command_has_no_failure_diagnostics(self):
runner = _runner_with_tool(_echo_tool())
mock_proc = MagicMock()
mock_proc.returncode = 0
mock_proc.stdout = "ok\n"
mock_proc.stderr = ""
with patch("subprocess.run", return_value=mock_proc):
result = runner.execute_tool("diagnostic_echo", {"msg": "hello"})
self.assertTrue(result.success)
self.assertTrue(hasattr(result, "error_code"))
self.assertIsNone(result.error_code)
self.assertIsNone(result.remediation)
if __name__ == "__main__":
unittest.main()