| 1 | """Sandbox backends for shell command execution. |
| 2 | |
| 3 | To add a new backend, implement a function with the signature: |
| 4 | _wrap_<name>(command: str, workspace: str, cwd: str) -> str |
| 5 | and register it in _BACKENDS below. |
| 6 | """ |
| 7 | |
| 8 | import shlex |
| 9 | from pathlib import Path |
| 10 | |
| 11 | from nanobot.config.paths import get_media_dir |
| 12 | |
| 13 | |
| 14 | def _bwrap(command: str, workspace: str, cwd: str) -> str: |
| 15 | """Wrap command in a bubblewrap sandbox (requires bwrap in container). |
| 16 | |
| 17 | Only the workspace is bind-mounted read-write; its parent dir (which holds |
| 18 | config.json) is hidden behind a fresh tmpfs. The media directory is |
| 19 | bind-mounted read-only so exec commands can read uploaded attachments. |
| 20 | """ |
| 21 | ws = Path(workspace).resolve() |
| 22 | media = get_media_dir().resolve() |
| 23 | |
| 24 | try: |
| 25 | sandbox_cwd = str(ws / Path(cwd).resolve().relative_to(ws)) |
| 26 | except ValueError: |
| 27 | sandbox_cwd = str(ws) |
| 28 | |
| 29 | required = ["/usr"] |
| 30 | optional = ["/bin", "/lib", "/lib64", "/etc/alternatives", |
| 31 | "/etc/ssl/certs", "/etc/resolv.conf", "/etc/ld.so.cache"] |
| 32 | |
| 33 | args = ["bwrap", "--new-session", "--die-with-parent"] |
| 34 | for p in required: args += ["--ro-bind", p, p] |
| 35 | for p in optional: args += ["--ro-bind-try", p, p] |
| 36 | args += [ |
| 37 | "--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp", |
| 38 | "--tmpfs", str(ws.parent), # mask config dir |
| 39 | "--dir", str(ws), # recreate workspace mount point |
| 40 | "--bind", str(ws), str(ws), |
| 41 | "--ro-bind-try", str(media), str(media), # read-only access to media |
| 42 | "--chdir", sandbox_cwd, |
| 43 | "--", "sh", "-c", command, |
| 44 | ] |
| 45 | return shlex.join(args) |
| 46 | |
| 47 | |
| 48 | _BACKENDS = {"bwrap": _bwrap} |
| 49 | |
| 50 | |
| 51 | def wrap_command(sandbox: str, command: str, workspace: str, cwd: str) -> str: |
| 52 | """Wrap *command* using the named sandbox backend.""" |
| 53 | if backend := _BACKENDS.get(sandbox): |
| 54 | return backend(command, workspace, cwd) |
| 55 | raise ValueError(f"Unknown sandbox backend {sandbox!r}. Available: {list(_BACKENDS)}") |
| 56 |