mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-10 22:38:40 +03:00
126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
"""Sandbox backends for shell command execution.
|
|
|
|
To add a new backend, implement a function with the signature:
|
|
_wrap_<name>(command: str, workspace: str, cwd: str) -> str
|
|
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 _normalize_bind_paths(
|
|
paths: Iterable[str] | None,
|
|
*,
|
|
workspace: Path | None = 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_path = path.resolve(strict=False)
|
|
if workspace is not None:
|
|
try:
|
|
workspace.relative_to(resolved_path)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
# A later bind of the workspace or one of its parents could
|
|
# cover the tmpfs that hides the config directory.
|
|
continue
|
|
resolved = str(resolved_path)
|
|
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
|
|
config.json) is hidden behind a fresh tmpfs. The media directory is
|
|
bind-mounted read-only so exec commands can read uploaded attachments.
|
|
"""
|
|
ws = Path(workspace).resolve()
|
|
media = get_media_dir().resolve()
|
|
|
|
try:
|
|
sandbox_cwd = str(ws / Path(cwd).resolve().relative_to(ws))
|
|
except ValueError:
|
|
sandbox_cwd = str(ws)
|
|
|
|
required = ["/usr"]
|
|
optional = [
|
|
"/bin",
|
|
"/lib",
|
|
"/lib64",
|
|
"/etc/alternatives",
|
|
"/etc/ssl/certs",
|
|
"/etc/pki/tls/certs",
|
|
"/etc/pki/ca-trust",
|
|
"/etc/crypto-policies",
|
|
"/etc/resolv.conf",
|
|
"/etc/ld.so.cache",
|
|
]
|
|
|
|
args = ["bwrap", "--new-session", "--die-with-parent", "--setenv", "HOME", str(ws)]
|
|
for p in required:
|
|
args += ["--ro-bind", p, p]
|
|
for p in optional:
|
|
args += ["--ro-bind-try", p, p]
|
|
args += [
|
|
"--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp",
|
|
"--tmpfs", str(ws.parent), # mask config dir
|
|
"--dir", str(ws), # recreate workspace mount point
|
|
"--bind", str(ws), str(ws),
|
|
"--ro-bind-try", str(media), str(media), # read-only access to media
|
|
]
|
|
for p in _normalize_bind_paths(sandbox_ro_binds, workspace=ws):
|
|
args += ["--ro-bind-try", p, p]
|
|
for p in _normalize_bind_paths(sandbox_rw_binds, workspace=ws):
|
|
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,
|
|
*,
|
|
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,
|
|
sandbox_ro_binds=sandbox_ro_binds,
|
|
sandbox_rw_binds=sandbox_rw_binds,
|
|
)
|
|
raise ValueError(f"Unknown sandbox backend {sandbox!r}. Available: {list(_BACKENDS)}")
|