feat(exec): allow extra bwrap bind roots

This commit is contained in:
yu-xin-c
2026-07-27 00:31:00 +08:00
committed by Xubin Ren
parent 5d8046deef
commit 01a11b3980
7 changed files with 263 additions and 6 deletions
+49 -5
View File
@@ -5,13 +5,40 @@ To add a new backend, implement a function with the signature:
and register it in _BACKENDS below.
"""
import os
import shlex
from pathlib import Path
from typing import Iterable
from nanobot.config.paths import get_media_dir
def _bwrap(command: str, workspace: str, cwd: str) -> str:
def _normalize_bind_paths(paths: Iterable[str] | None) -> list[str]:
out: list[str] = []
seen: set[str] = set()
for raw in paths or []:
value = str(raw).strip()
if not value:
continue
path = Path(os.path.expandvars(value)).expanduser()
if not path.is_absolute():
continue
resolved = str(path.resolve(strict=False))
if resolved in seen:
continue
seen.add(resolved)
out.append(resolved)
return out
def _bwrap(
command: str,
workspace: str,
cwd: str,
*,
sandbox_ro_binds: Iterable[str] | None = None,
sandbox_rw_binds: Iterable[str] | None = None,
) -> str:
"""Wrap command in a bubblewrap sandbox (requires bwrap in container).
Only the workspace is bind-mounted read-write; its parent dir (which holds
@@ -51,17 +78,34 @@ def _bwrap(command: str, workspace: str, cwd: str) -> str:
"--dir", str(ws), # recreate workspace mount point
"--bind", str(ws), str(ws),
"--ro-bind-try", str(media), str(media), # read-only access to media
"--chdir", sandbox_cwd,
"--", "sh", "-c", command,
]
for p in _normalize_bind_paths(sandbox_ro_binds):
args += ["--ro-bind-try", p, p]
for p in _normalize_bind_paths(sandbox_rw_binds):
args += ["--bind-try", p, p]
args += ["--chdir", sandbox_cwd, "--", "sh", "-c", command]
return shlex.join(args)
_BACKENDS = {"bwrap": _bwrap}
def wrap_command(sandbox: str, command: str, workspace: str, cwd: str) -> str:
def wrap_command(
sandbox: str,
command: str,
workspace: str,
cwd: str,
*,
sandbox_ro_binds: Iterable[str] | None = None,
sandbox_rw_binds: Iterable[str] | None = None,
) -> str:
"""Wrap *command* using the named sandbox backend."""
if backend := _BACKENDS.get(sandbox):
return backend(command, workspace, cwd)
return backend(
command,
workspace,
cwd,
sandbox_ro_binds=sandbox_ro_binds,
sandbox_rw_binds=sandbox_rw_binds,
)
raise ValueError(f"Unknown sandbox backend {sandbox!r}. Available: {list(_BACKENDS)}")
+44 -1
View File
@@ -84,6 +84,8 @@ class ExecToolConfig(Base):
path_prepend: str = ""
path_append: str = ""
sandbox: str = ""
sandbox_ro_binds: list[str] = Field(default_factory=list)
sandbox_rw_binds: list[str] = Field(default_factory=list)
allowed_env_keys: list[str] = Field(default_factory=list)
allow_patterns: list[str] = Field(default_factory=list)
deny_patterns: list[str] = Field(default_factory=list)
@@ -187,6 +189,8 @@ class ExecTool(Tool):
sandbox=cfg.sandbox,
path_prepend=cfg.path_prepend,
path_append=cfg.path_append,
sandbox_ro_binds=cfg.sandbox_ro_binds,
sandbox_rw_binds=cfg.sandbox_rw_binds,
allowed_env_keys=cfg.allowed_env_keys,
allow_patterns=cfg.allow_patterns,
deny_patterns=cfg.deny_patterns,
@@ -205,6 +209,8 @@ class ExecTool(Tool):
sandbox: str = "",
path_prepend: str = "",
path_append: str = "",
sandbox_ro_binds: list[str] | None = None,
sandbox_rw_binds: list[str] | None = None,
allowed_env_keys: list[str] | None = None,
session_manager: Any | None = None,
):
@@ -237,6 +243,8 @@ class ExecTool(Tool):
self.webui_allow_local_service_access = webui_allow_local_service_access
self.path_prepend = path_prepend
self.path_append = path_append
self.sandbox_ro_binds = self._normalize_bind_roots(sandbox_ro_binds)
self.sandbox_rw_binds = self._normalize_bind_roots(sandbox_rw_binds)
self.allowed_env_keys = allowed_env_keys or []
self._session_manager = session_manager or DEFAULT_EXEC_SESSION_MANAGER
@@ -464,7 +472,14 @@ class ExecTool(Tool):
)
else:
workspace = workspace_root or cwd
command = wrap_command(self.sandbox, command, workspace, cwd)
command = wrap_command(
self.sandbox,
command,
workspace,
cwd,
sandbox_ro_binds=[str(p) for p in self.sandbox_ro_binds],
sandbox_rw_binds=[str(p) for p in self.sandbox_rw_binds],
)
cwd = str(Path(workspace).resolve())
effective_timeout = self._resolve_timeout(timeout)
@@ -794,6 +809,7 @@ class ExecTool(Tool):
if workspace_root
else None
)
sandbox_bind_roots = self._active_sandbox_bind_roots()
for raw in self._extract_absolute_paths(cmd):
try:
@@ -817,6 +833,8 @@ class ExecTool(Tool):
)
if not allowed and resolved_workspace is not None:
allowed = is_path_within(p, resolved_workspace)
if not allowed and sandbox_bind_roots:
allowed = any(is_path_within(p, root) for root in sandbox_bind_roots)
if p.is_absolute() and not allowed:
return ToolResult.error(
"Error: Command blocked by safety guard (path outside working dir)"
@@ -921,3 +939,28 @@ class ExecTool(Tool):
posix_paths = re.findall(r"(?:^|[\s|>='\"])(/[^\s\"'>;|<]+)", command) # POSIX: /absolute only
home_paths = re.findall(r"(?:^|[\s>='\"])(~[/+][^\s\"'>;|<]*)", command) # POSIX/Windows home shortcut: ~/ or ~+
return win_paths + posix_paths + home_paths
@staticmethod
def _normalize_bind_roots(paths: list[str] | None) -> list[Path]:
roots: list[Path] = []
seen: set[str] = set()
for raw in paths or []:
value = str(raw).strip()
if not value:
continue
path = Path(os.path.expandvars(value)).expanduser()
if not path.is_absolute():
continue
with suppress(OSError, RuntimeError, ValueError):
resolved = path.resolve(strict=False)
key = os.path.normcase(os.fspath(resolved))
if key in seen:
continue
seen.add(key)
roots.append(resolved)
return roots
def _active_sandbox_bind_roots(self) -> list[Path]:
if self.sandbox != "bwrap" or _IS_WINDOWS:
return []
return [*self.sandbox_ro_binds, *self.sandbox_rw_binds]