Something went wrong. Try again.
Embedded bash sandbox for agents, inspired by gVisor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493"""cVisor Python SDK — a thin ctypes wrapper over the libcvisor C ABI.
Example: from cvisor import Sandbox out = Sandbox().run("echo hello") print(out.stdout) # "hello\n""""
from __future__ import annotations
import ctypesimport osimport platformimport threadingimport timefrom ctypes import ( c_char_p, c_int, c_long, c_size_t, c_uint8, c_uint16, c_uint32, c_uint64, c_void_p, POINTER,)
from .graphql import GraphQLClient, GraphQLErrorfrom .remote import RemoteSandbox
__all__ = [ "Sandbox", "Output", "Session", "load_library", "GraphQLClient", "GraphQLError", "RemoteSandbox",]
def _default_library_path() -> str: """Resolve libcvisor.so: CVISOR_LIB env override, else a bundled copy.""" override = os.environ.get("CVISOR_LIB") if override: return override here = os.path.dirname(os.path.abspath(__file__)) arch = "aarch64" if platform.machine() in ("aarch64", "arm64") else "x86_64" candidates = [ os.path.join(here, "_native", f"libcvisor-{arch}.so"), os.path.join(here, "_native", "libcvisor.so"), os.path.join(here, "libcvisor.so"), ] for c in candidates: if os.path.exists(c): return c # Fall back to the first candidate so the error names a concrete path. return candidates[0]
def load_library(path: str | None = None) -> ctypes.CDLL: lib = ctypes.CDLL(path or _default_library_path())
lib.cvisor_sandbox_new.restype = c_void_p lib.cvisor_sandbox_new.argtypes = []
lib.cvisor_sandbox_free.restype = None lib.cvisor_sandbox_free.argtypes = [c_void_p]
lib.cvisor_sandbox_set_log_level.restype = None lib.cvisor_sandbox_set_log_level.argtypes = [c_void_p, c_int]
lib.cvisor_sandbox_set_allow_network.restype = None lib.cvisor_sandbox_set_allow_network.argtypes = [c_void_p, c_int]
lib.cvisor_sandbox_set_allow_listen.restype = None lib.cvisor_sandbox_set_allow_listen.argtypes = [c_void_p, c_int]
lib.cvisor_sandbox_set_env.restype = None lib.cvisor_sandbox_set_env.argtypes = [c_void_p, c_char_p, c_char_p]
lib.cvisor_sandbox_set_limits.restype = None lib.cvisor_sandbox_set_limits.argtypes = [c_void_p, c_uint64, c_uint64, c_uint32]
lib.cvisor_sandbox_write_file.restype = c_int lib.cvisor_sandbox_write_file.argtypes = [c_void_p, c_char_p, POINTER(c_uint8), c_size_t]
lib.cvisor_sandbox_read_file.restype = POINTER(c_uint8) lib.cvisor_sandbox_read_file.argtypes = [c_void_p, c_char_p, POINTER(c_size_t)]
lib.cvisor_sandbox_copy_into.restype = c_int lib.cvisor_sandbox_copy_into.argtypes = [c_void_p, c_char_p, c_char_p]
lib.cvisor_sandbox_copy_out.restype = c_int lib.cvisor_sandbox_copy_out.argtypes = [c_void_p, c_char_p, c_char_p]
lib.cvisor_cache_save.restype = c_int lib.cvisor_cache_save.argtypes = [c_void_p, c_char_p, c_char_p, c_char_p, c_char_p]
lib.cvisor_cache_restore.restype = c_int lib.cvisor_cache_restore.argtypes = [c_void_p, c_char_p, c_char_p, c_char_p, c_char_p]
lib.cvisor_run.restype = c_void_p lib.cvisor_run.argtypes = [c_void_p, c_char_p]
lib.cvisor_run_timeout.restype = c_void_p lib.cvisor_run_timeout.argtypes = [c_void_p, c_char_p, c_uint64]
lib.cvisor_output_free.restype = None lib.cvisor_output_free.argtypes = [c_void_p]
lib.cvisor_output_exit_code.restype = c_int lib.cvisor_output_exit_code.argtypes = [c_void_p]
for fn in ("cvisor_output_stdout", "cvisor_output_stderr"): f = getattr(lib, fn) f.restype = POINTER(c_uint8) f.argtypes = [c_void_p, POINTER(c_size_t)]
lib.cvisor_bytes_free.restype = None lib.cvisor_bytes_free.argtypes = [POINTER(c_uint8), c_size_t]
lib.cvisor_session_start.restype = c_void_p lib.cvisor_session_start.argtypes = [c_void_p, c_char_p, c_int]
for fn in ("cvisor_session_read_stdout", "cvisor_session_read_stderr"): f = getattr(lib, fn) f.restype = POINTER(c_uint8) f.argtypes = [c_void_p, POINTER(c_size_t)]
lib.cvisor_session_write_stdin.restype = c_long lib.cvisor_session_write_stdin.argtypes = [c_void_p, POINTER(c_uint8), c_size_t]
lib.cvisor_session_resize.restype = None lib.cvisor_session_resize.argtypes = [c_void_p, c_uint16, c_uint16]
lib.cvisor_session_try_wait.restype = c_int lib.cvisor_session_try_wait.argtypes = [c_void_p, POINTER(c_int)]
lib.cvisor_session_wait.restype = c_int lib.cvisor_session_wait.argtypes = [c_void_p]
lib.cvisor_session_kill.restype = None lib.cvisor_session_kill.argtypes = [c_void_p]
lib.cvisor_session_free.restype = None lib.cvisor_session_free.argtypes = [c_void_p]
return lib
_LIB: ctypes.CDLL | None = None
def _lib() -> ctypes.CDLL: global _LIB if _LIB is None: _LIB = load_library() return _LIB
class Output: """Captured output of one sandbox run."""
def __init__(self, stdout: bytes, stderr: bytes, exit_code: int) -> None: self.stdout_bytes = stdout self.stderr_bytes = stderr self.exit_code = exit_code
@property def stdout(self) -> str: return self.stdout_bytes.decode("utf-8", "replace")
@property def stderr(self) -> str: return self.stderr_bytes.decode("utf-8", "replace")
class Session: """A live, streaming sandbox session.
Wraps an opaque ``CvisorSession*``. For a PTY session stdout and stderr are merged (drain stdout; stderr is empty) and stdin can be written; for a plain session the two streams are separate and stdin is unavailable. """
def __init__(self, lib: ctypes.CDLL, ptr: int) -> None: self._lib = lib self._ptr = ptr
def _read(self, accessor) -> bytes: n = c_size_t(0) ptr = accessor(self._ptr, ctypes.byref(n)) if not ptr or n.value == 0: return b"" try: return bytes(ctypes.cast(ptr, POINTER(c_uint8 * n.value)).contents) finally: self._lib.cvisor_bytes_free(ptr, n.value)
def read_stdout(self) -> bytes: return self._read(self._lib.cvisor_session_read_stdout)
def read_stderr(self) -> bytes: return self._read(self._lib.cvisor_session_read_stderr)
def write_stdin(self, data: bytes | str) -> int: if isinstance(data, str): data = data.encode("utf-8") buf = (c_uint8 * len(data)).from_buffer_copy(data) return self._lib.cvisor_session_write_stdin(self._ptr, buf, len(data))
def resize(self, rows: int, cols: int) -> None: self._lib.cvisor_session_resize(self._ptr, rows, cols)
def exit_code(self) -> int | None: """Return the exit code if the session has finished, else None.""" done = c_int(0) code = self._lib.cvisor_session_try_wait(self._ptr, ctypes.byref(done)) return code if done.value else None
def wait(self) -> int: return self._lib.cvisor_session_wait(self._ptr)
def kill(self) -> None: self._lib.cvisor_session_kill(self._ptr)
def close(self) -> None: if getattr(self, "_ptr", None): self._lib.cvisor_session_free(self._ptr) self._ptr = None
def __del__(self) -> None: self.close()
def __enter__(self) -> "Session": return self
def __exit__(self, *exc) -> None: self.close()
class Sandbox: def __init__(self) -> None: if platform.system() != "Linux": raise RuntimeError( "the local FFI Sandbox is Linux-only; use the GraphQL client " "(cvisor.RemoteSandbox / cvisor.GraphQLClient) on this platform" ) self._lib = _lib() self._ptr = self._lib.cvisor_sandbox_new() if not self._ptr: raise RuntimeError("failed to create sandbox")
def set_log_level(self, level: str) -> None: self._lib.cvisor_sandbox_set_log_level(self._ptr, 1 if level == "DEBUG" else 0)
def set_allow_network(self, allow: bool) -> None: """Allow (default) or deny outbound INET/INET6 networking.""" self._lib.cvisor_sandbox_set_allow_network(self._ptr, 1 if allow else 0)
def set_allow_listen(self, allow: bool) -> None: """Allow or deny (default) inbound TCP servers (listen/accept).""" self._lib.cvisor_sandbox_set_allow_listen(self._ptr, 1 if allow else 0)
def set_env(self, key: str, value: str) -> None: """Set an environment variable for the guest (applies to later runs).""" self._lib.cvisor_sandbox_set_env( self._ptr, key.encode("utf-8"), value.encode("utf-8") )
def set_limits( self, memory_max: int = 0, pids_max: int = 0, cpu_percent: int = 0, ) -> None: """Cap guest resources via cgroup v2 (applies to later runs).
``memory_max`` is bytes, ``pids_max`` a process count, ``cpu_percent`` percent of one core (50 = half a core). 0 leaves that limit unset.""" self._lib.cvisor_sandbox_set_limits( self._ptr, memory_max, pids_max, cpu_percent )
def write_file(self, path: str, data: bytes | str) -> None: """Write ``data`` to ``path`` in the sandbox's persistent overlay.
``path`` must be absolute; ``/proc`` and passthrough paths are not writable. The file is visible to later ``run`` calls of this same Sandbox instance. Raises ``OSError`` on failure.""" if isinstance(data, str): data = data.encode("utf-8") buf = (c_uint8 * len(data)).from_buffer_copy(data) rc = self._lib.cvisor_sandbox_write_file( self._ptr, path.encode("utf-8"), buf, len(data) ) if rc != 0: raise OSError(-rc, os.strerror(-rc), path)
def read_file(self, path: str) -> bytes: """Read ``path`` as the guest sees it (overlay copy, else the real host file for cow paths) and return its bytes.
``path`` must be absolute. An empty file returns ``b""``; a missing or unreadable path raises ``FileNotFoundError``.""" n = c_size_t(0) ptr = self._lib.cvisor_sandbox_read_file( self._ptr, path.encode("utf-8"), ctypes.byref(n) ) if not ptr: if n.value == 0: return b"" raise FileNotFoundError(path) try: return bytes(ctypes.cast(ptr, POINTER(c_uint8 * n.value)).contents) finally: self._lib.cvisor_bytes_free(ptr, n.value)
def copy_into(self, host_path: str, guest_path: str) -> None: """Copy a host file or directory tree into the sandbox overlay.
``host_path`` may be a single file or a directory; directories are copied recursively and respect ``.gitignore``/``.dockerignore``. The copied files land at ``guest_path`` in the sandbox's persistent overlay and are visible to later ``run`` calls. Raises ``OSError`` on failure.""" rc = self._lib.cvisor_sandbox_copy_into( self._ptr, host_path.encode("utf-8"), guest_path.encode("utf-8") ) if rc != 0: raise OSError(-rc, os.strerror(-rc), host_path)
def copy_out(self, guest_path: str, host_path: str) -> None: """Copy a file or directory tree out of the sandbox overlay to the host.
``guest_path`` is resolved as the guest sees it and written to ``host_path`` on the real filesystem. Raises ``OSError`` on failure.""" rc = self._lib.cvisor_sandbox_copy_out( self._ptr, guest_path.encode("utf-8"), host_path.encode("utf-8") ) if rc != 0: raise OSError(-rc, os.strerror(-rc), host_path)
def cache_save( self, sandbox_path: str, key: str, backend: str = "", format: str = "gzip" ) -> None: """Archive a sandbox directory under ``key`` to a cache backend.
``sandbox_path`` is a directory in the sandbox; its contents (respecting ``.gitignore``/``.dockerignore``) are archived under ``key``.
``backend`` selects where the archive is stored: ``""`` or ``"disk"`` (host disk, the default), ``"disk:/path"``, or ``"s3://bucket/prefix?region=..&endpoint=.."``. S3 requires the library to be built with the ``s3`` feature; the bundled ``libcvisor.so`` is not, so ``disk`` is the working default.
``format`` selects the archive format: ``""`` or ``"gzip"`` (the default), ``"estargz"``, ``"none"`` (uncompressed), or ``"zstd"`` (only if the library is built with zstd; the bundled lib is not).
Raises ``OSError`` on failure.""" rc = self._lib.cvisor_cache_save( self._ptr, sandbox_path.encode("utf-8"), key.encode("utf-8"), backend.encode("utf-8"), format.encode("utf-8"), ) if rc != 0: raise OSError(-rc, os.strerror(-rc), key)
def cache_restore( self, sandbox_path: str, key: str, backend: str = "", format: str = "gzip" ) -> None: """Restore a cached archive stored under ``key`` back into the overlay.
Unpacks the archive saved by :meth:`cache_save` into ``sandbox_path`` in the sandbox's persistent overlay, so its files are visible to later ``run`` calls. ``backend`` and ``format`` must match what was used to save (see :meth:`cache_save` for the accepted values and the disk-only note for the bundled library). Raises ``OSError`` on failure.""" rc = self._lib.cvisor_cache_restore( self._ptr, sandbox_path.encode("utf-8"), key.encode("utf-8"), backend.encode("utf-8"), format.encode("utf-8"), ) if rc != 0: raise OSError(-rc, os.strerror(-rc), key)
def run(self, command: str, timeout_ms: int | None = None) -> Output: """Run a shell command; if timeout_ms is set, SIGKILL the guest after that many milliseconds (a timed-out run reports exit code 137).""" cmd = command.encode("utf-8") if timeout_ms is not None and timeout_ms > 0: out = self._lib.cvisor_run_timeout(self._ptr, cmd, timeout_ms) else: out = self._lib.cvisor_run(self._ptr, cmd) if not out: raise RuntimeError("sandbox run failed") try: return Output(self._read(out, self._lib.cvisor_output_stdout), self._read(out, self._lib.cvisor_output_stderr), self._lib.cvisor_output_exit_code(out)) finally: self._lib.cvisor_output_free(out)
def _read(self, out: int, accessor) -> bytes: n = c_size_t(0) ptr = accessor(out, ctypes.byref(n)) if not ptr or n.value == 0: return b"" try: return bytes(ctypes.cast(ptr, POINTER(c_uint8 * n.value)).contents) finally: self._lib.cvisor_bytes_free(ptr, n.value)
def run_streaming(self, command: str, on_stdout=None, on_stderr=None, poll_ms: int = 15) -> int: """Run a command, streaming output to callbacks as it is produced.
Starts a non-PTY session and polls it: whenever new stdout/stderr bytes arrive they are decoded (utf-8, errors="replace") and passed to the matching callback. Returns the command's exit code.""" ptr = self._lib.cvisor_session_start(self._ptr, command.encode("utf-8"), 0) if not ptr: raise RuntimeError("failed to start session") session = Session(self._lib, ptr) try: while True: out = session.read_stdout() if out and on_stdout is not None: on_stdout(out.decode("utf-8", "replace")) err = session.read_stderr() if err and on_stderr is not None: on_stderr(err.decode("utf-8", "replace")) code = session.exit_code() if code is not None: out = session.read_stdout() if out and on_stdout is not None: on_stdout(out.decode("utf-8", "replace")) err = session.read_stderr() if err and on_stderr is not None: on_stderr(err.decode("utf-8", "replace")) return code time.sleep(poll_ms / 1000) finally: session.close()
def shell(self, on_output=None, poll_ms: int = 15) -> Session: """Start an interactive PTY shell (/bin/sh -i) as a streaming session.
If ``on_output`` is given, a daemon thread drains merged output and passes each decoded chunk (utf-8, errors="replace") to it until the shell exits. Returns the Session so the caller can write_stdin/resize/ wait/close it.""" ptr = self._lib.cvisor_session_start(self._ptr, None, 1) if not ptr: raise RuntimeError("failed to start session") session = Session(self._lib, ptr) if on_output is not None: def pump() -> None: while True: out = session.read_stdout() if out: on_output(out.decode("utf-8", "replace")) if session.exit_code() is not None: out = session.read_stdout() if out: on_output(out.decode("utf-8", "replace")) return time.sleep(poll_ms / 1000)
threading.Thread(target=pump, daemon=True).start() return session
def close(self) -> None: if getattr(self, "_ptr", None): self._lib.cvisor_sandbox_free(self._ptr) self._ptr = None
def __del__(self) -> None: self.close()
def __enter__(self) -> "Sandbox": return self
def __exit__(self, *exc) -> None: self.close()