sand-boxer/src/sandboxer/extensions/bwrap.py

"""ext.bwrap — local, same-host, bubblewrap namespace isolation (SAND-WP-0013)."""

from __future__ import annotations

import os
import shutil
import signal
import subprocess
from pathlib import Path
from typing import Any

from sandboxer.extensions.base import SandboxExtension
from sandboxer.models import Profile


class BwrapExtension(SandboxExtension):
    """Provision a local sandbox via bubblewrap (bwrap) kernel namespaces.

    Unlike ext.compose-ssh / ext.vm-packer, this extension never leaves the
    local host: no SSH hop, no container runtime, no remote placement. A new
    user/mount/pid/ipc/uts/net namespace is created per sandbox, kept alive
    by a long-running placeholder process (`sleep infinity`) whose pid is
    the handle's exec target. `--unshare-net` with no veth/interface makes
    `network.default: deny` real, rather than declarative-only like the
    other self-hosted extensions.
    """

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        super().__init__(config)
        cfg = self.config
        self.base_dir: str = cfg.get("base_dir", "/tmp/sandboxer-bwrap")
        self.bwrap_bin: str = cfg.get("bwrap_bin", "bwrap")
        self.ro_binds: list[str] = cfg.get(
            "ro_binds", ["/usr", "/bin", "/lib", "/lib64", "/etc/resolv.conf"]
        )

    def _bwrap_bin(self) -> str:
        return os.environ.get("SANDBOXER_BWRAP_BIN", self.bwrap_bin)

    def _existing_ro_binds(self) -> list[str]:
        return [path for path in self.ro_binds if Path(path).exists()]

    def _bwrap_argv(self, workspace_dir: str) -> list[str]:
        argv = [
            self._bwrap_bin(),
            "--die-with-parent",
            "--unshare-user",
            "--unshare-pid",
            "--unshare-ipc",
            "--unshare-uts",
            "--unshare-cgroup",
            "--unshare-net",
            "--tmpfs",
            "/",
            "--proc",
            "/proc",
            "--dev",
            "/dev",
        ]
        for path in self._existing_ro_binds():
            argv += ["--ro-bind", path, path]
        argv += ["--bind", workspace_dir, workspace_dir]
        argv += ["--chdir", workspace_dir]
        argv += ["sleep", "infinity"]
        return argv

    def provision(
        self, profile: Profile, inputs: dict[str, str], host: str
    ) -> dict[str, str]:
        sandbox_id = self.new_sandbox_id(inputs)
        workspace_dir = f"{self.base_dir}/{sandbox_id}"
        Path(workspace_dir).mkdir(parents=True, exist_ok=True)

        repo = inputs.get("repo")
        if repo:
            repo_path = Path(repo).expanduser().resolve()
            if not repo_path.exists():
                raise FileNotFoundError(f"Repo path does not exist: {repo_path}")
            shutil.copytree(repo_path, workspace_dir, dirs_exist_ok=True)

        argv = self._bwrap_argv(workspace_dir)
        proc = subprocess.Popen(
            argv,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )

        return {
            "sandbox_id": sandbox_id,
            "host": host,
            "pid": str(proc.pid),
            "workspace_dir": workspace_dir,
        }

    def wait_ready(self, handle: dict[str, str]) -> dict[str, str]:
        pid = int(handle["pid"])
        if not self._pid_alive(pid):
            raise RuntimeError(f"bwrap process {pid} is not running")
        workspace_dir = handle["workspace_dir"]
        if not Path(workspace_dir).is_dir():
            raise RuntimeError(f"workspace missing: {workspace_dir}")
        return {
            "host": handle.get("host", "localhost"),
            "endpoint": f"pid:{pid}",
        }

    def teardown(self, handle: dict[str, str]) -> dict[str, str]:
        pid_str = handle.get("pid", "")
        killed = False
        if pid_str and self._pid_alive(int(pid_str)):
            pid = int(pid_str)
            try:
                os.killpg(os.getpgid(pid), signal.SIGKILL)
            except (ProcessLookupError, PermissionError):
                try:
                    os.kill(pid, signal.SIGKILL)
                except (ProcessLookupError, PermissionError):
                    pass
            killed = True

        workspace_dir = handle.get("workspace_dir", "")
        removed = False
        if workspace_dir and Path(workspace_dir).exists():
            shutil.rmtree(workspace_dir, ignore_errors=True)
            removed = not Path(workspace_dir).exists()

        return {
            "process_killed": str(killed),
            "workspace_removed": str(removed),
            "workspace_dir": workspace_dir,
        }

    @staticmethod
    def _pid_alive(pid: int) -> bool:
        try:
            os.kill(pid, 0)
        except (ProcessLookupError, PermissionError):
            return False
        return True