diff --git a/sandboxexec/sandbox/BUILD b/sandboxexec/sandbox/BUILD index b66faec694..adc566f923 100644 --- a/sandboxexec/sandbox/BUILD +++ b/sandboxexec/sandbox/BUILD @@ -1,3 +1,5 @@ +# Placeholder: load py_library +# Placeholder: load py_test load("//tools:defs.bzl", "go_library", "go_test") package( @@ -29,3 +31,20 @@ go_test( "@com_github_opencontainers_runtime_spec//specs-go:go_default_library", ], ) + +py_library( + name = "sandbox_py", + srcs = ["sandbox.py"], + visibility = ["//:__subpackages__"], +) + +py_test( + name = "sandbox_py_test", + srcs = ["sandbox_test.py"], + data = ["//runsc"], + main = "sandbox_test.py", + tags = ["nogotsan"], + deps = [ + ":sandbox_py", + ], +) diff --git a/sandboxexec/sandbox/sandbox.py b/sandboxexec/sandbox/sandbox.py new file mode 100644 index 0000000000..0a66258ba6 --- /dev/null +++ b/sandboxexec/sandbox/sandbox.py @@ -0,0 +1,310 @@ +# Copyright 2026 The gVisor Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Python binding for gVisor sandbox. + +This module provides a Python API for creating gVisor sandboxes and executing +commands inside them. +""" + +import json +import os +import random +import shutil +import subprocess +import tempfile +from typing import Optional, Tuple + + +class SandboxError(Exception): + """Base exception for Sandbox operations.""" + + +class Sandbox: + """Represents a running gVisor sandbox.""" + + def __init__( + self, + runtime_dir: Optional[str] = None, + sandbox_id: Optional[str] = None, + enable_networking: bool = True, + ): + """Initializes and starts a new sandbox. + + Args: + runtime_dir: Custom runtime directory where bundle and state files are + written. If not set, a temporary directory is created. + sandbox_id: Specific sandbox ID. If not set, a unique ID is generated + automatically. + enable_networking: Whether networking is enabled inside the sandbox. + + Raises: + SandboxError: If sandbox creation fails. + """ + self._enable_networking = enable_networking + self._runtime_dir = "" + self._owns_runtime_dir = False + self._id = "" + self._state_dir = "" + self._bundle_dir = "" + self._closed = False + self._runsc_path = self._find_runsc() + + if runtime_dir is None: + try: + self._runtime_dir = tempfile.mkdtemp(prefix="gvisor-sandbox-") + except OSError as e: + raise SandboxError(f"failed to create runtime directory: {e}") from e + self._owns_runtime_dir = True + else: + self._runtime_dir = runtime_dir + self._owns_runtime_dir = False + + self._id = sandbox_id or self._generate_id() + + try: + if os.geteuid() != 0 and self._enable_networking: + raise SandboxError("enabling networking requires running as root") + + self._state_dir = os.path.join(self._runtime_dir, "state") + try: + os.makedirs(self._state_dir, mode=0o700, exist_ok=True) + except OSError as e: + raise SandboxError( + f"failed to create sandbox state directory: {e}" + ) from e + + # Verify permissions (mode might be different if directory already existed). + try: + stat_info = os.stat(self._state_dir) + if (stat_info.st_mode & 0o777) != 0o700: + os.chmod(self._state_dir, 0o700) + except OSError as e: + raise SandboxError( + "sandbox state directory has incorrect permissions and failed to" + f" chmod: {e}" + ) from e + + self._bundle_dir = self._create_bundle() + + # Launch the sandbox in detached mode. + args = ["--root", self._state_dir] + if os.geteuid() != 0: + args.append("--ignore-cgroups") + if not self._enable_networking: + args.append("--network=none") + args.extend(["run", "--bundle", self._bundle_dir, "--detach", self._id]) + + # We must use a file for stderr because runsc run with --detach spawns a + # grandchild process. If we use pipes (e.g. capture_output=True), Python's + # subprocess.run will hang waiting for the pipes to close. Writing to a + # file avoids this hang while still allowing us to capture errors. + stderr_path = os.path.join(self._runtime_dir, "runsc-stderr.log") + try: + with open(stderr_path, "w+b") as stderr_file: + subprocess.run( + [self._runsc_path] + args, + check=True, + stdout=subprocess.DEVNULL, + stderr=stderr_file, + timeout=30, + ) + except subprocess.TimeoutExpired as e: + raise SandboxError("sandbox creation timed out (runsc run hung)") from e + except subprocess.CalledProcessError as e: + stderr_content = "" + if os.path.exists(stderr_path): + try: + with open(stderr_path, "r", errors="replace") as f: + stderr_content = f.read() + except OSError: + pass + raise SandboxError( + f"failed to create sandbox via subprocess: exit code {e.returncode}," + f" stderr: {stderr_content}" + ) from e + except Exception: + self.close() + raise + + def __enter__(self) -> "Sandbox": + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + @property + def id(self) -> str: + """Returns the sandbox ID.""" + return self._id + + @property + def bundle_dir(self) -> str: + """Returns the path to the OCI bundle directory.""" + return self._bundle_dir + + def _generate_id(self) -> str: + return f"{random.getrandbits(128):032x}" + + def _find_runsc(self) -> str: + if "RUNSC_PATH" in os.environ: + return os.environ["RUNSC_PATH"] + path = shutil.which("runsc") + if path is not None: + return path + raise SandboxError("runsc binary is not found") + + def _create_bundle(self) -> str: + """Creates the OCI bundle directory and config.json. + + Returns: + The path to the created bundle directory. + + Raises: + SandboxError: If bundle creation fails. + """ + bundle_dir = os.path.join(self._runtime_dir, self._id) + rootfs_dir = os.path.join(bundle_dir, "rootfs") + try: + os.makedirs(rootfs_dir, mode=0o755, exist_ok=True) + except OSError as e: + raise SandboxError(f"failed to create bundle directories: {e}") from e + + namespaces = [ + {"type": "pid"}, + {"type": "mount"}, + {"type": "uts"}, + {"type": "ipc"}, + ] + if os.geteuid() != 0: + namespaces.append({"type": "user"}) + if self._enable_networking: + namespaces.append({"type": "network"}) + + mounts = [ + {"destination": "/proc", "type": "proc", "source": "proc"}, + {"destination": "/dev", "type": "tmpfs", "source": "tmpfs"}, + ] + + for p in ["/bin", "/usr", "/lib", "/lib64", "/etc/alternatives"]: + if os.path.exists(p): + opts = ["rbind", "ro", "nosuid", "nodev"] + if p == "/etc/alternatives": + opts = ["rbind", "ro"] + mounts.append({ + "destination": p, + "type": "bind", + "source": p, + "options": opts, + }) + + linux = { + "namespaces": namespaces, + } + if os.geteuid() != 0: + linux["uidMappings"] = [ + {"containerID": 0, "hostID": os.geteuid(), "size": 1} + ] + linux["gidMappings"] = [ + {"containerID": 0, "hostID": os.getegid(), "size": 1} + ] + + spec = { + "ociVersion": "1.0.0", + "root": { + "path": "rootfs", + "readonly": True, + }, + "process": { + "terminal": False, + "user": {"uid": 0, "gid": 0}, + "args": ["sleep", "infinity"], + "cwd": "/", + "env": ["PATH=/bin:/usr/bin:/usr/local/bin"], + }, + "mounts": mounts, + "linux": linux, + } + + config_path = os.path.join(bundle_dir, "config.json") + try: + with open(config_path, "w") as f: + json.dump(spec, f, indent=2) + except OSError as e: + raise SandboxError(f"failed to create config.json: {e}") from e + + return bundle_dir + + def exec( + self, cmd: str, *args: str, timeout: Optional[float] = None + ) -> Tuple[str, str]: + """Runs the given command inside the running sandbox. + + Args: + cmd: The command to run. + *args: Arguments to the command. + timeout: Timeout in seconds. + + Returns: + A tuple of (stdout, stderr) strings. + + Raises: + SandboxError: If the command execution fails or times out. + """ + runsc_args = ["--root", self._state_dir, "exec", self._id, cmd] + list(args) + try: + result = subprocess.run( + [self._runsc_path] + runsc_args, + capture_output=True, + text=True, + check=True, + timeout=timeout, + ) + return result.stdout, result.stderr + except subprocess.TimeoutExpired as e: + raise SandboxError(f"exec timed out after {timeout} seconds") from e + except subprocess.CalledProcessError as e: + raise SandboxError(f"exec failed: {e.stderr}") from e + + def close(self): + """Kills the sandbox processes and cleans up directories.""" + if self._closed: + return + self._closed = True + + if self._state_dir: + kill_args = ["--root", self._state_dir, "kill", self._id, "SIGKILL"] + subprocess.run( + [self._runsc_path] + kill_args, capture_output=True, check=False + ) + + delete_args = ["--root", self._state_dir, "delete", "--force", self._id] + delete_result = subprocess.run( + [self._runsc_path] + delete_args, capture_output=True, check=False + ) + if delete_result.returncode != 0: + # We might want to log this, but we continue cleanup anyway. + pass + + if self._bundle_dir and os.path.exists(self._bundle_dir): + try: + shutil.rmtree(self._bundle_dir) + except OSError: + pass + + if self._owns_runtime_dir and os.path.exists(self._runtime_dir): + try: + shutil.rmtree(self._runtime_dir) + except OSError: + pass diff --git a/sandboxexec/sandbox/sandbox_test.py b/sandboxexec/sandbox/sandbox_test.py new file mode 100644 index 0000000000..5e9aab4e09 --- /dev/null +++ b/sandboxexec/sandbox/sandbox_test.py @@ -0,0 +1,196 @@ +# Copyright 2026 The gVisor Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the gVisor sandbox Python binding.""" + +import glob +import json +import os +import shutil +import tempfile +import unittest + +from sandboxexec.sandbox import sandbox + + +def find_runsc() -> str: + """Finds the runsc binary in the test environment.""" + if "RUNSC_PATH" in os.environ: + return os.environ["RUNSC_PATH"] + + # Try to find via runfiles directory recursively. + for env_var in ["RUNFILES_DIR", "PYTHON_RUNFILES"]: + runfiles_dir = os.environ.get(env_var) + if runfiles_dir: + pattern = os.path.join(runfiles_dir, "**", "runsc/runsc") + matches = glob.glob(pattern, recursive=True) + matches = [m for m in matches if os.path.isfile(m)] + if matches: + return matches[0] + + # Fallback to PATH. + path = shutil.which("runsc") + if path: + return path + + raise RuntimeError("runsc binary not found") + + +def setUpModule(): + try: + os.environ["RUNSC_PATH"] = find_runsc() + except RuntimeError as e: + raise unittest.SkipTest(str(e)) + + +class SandboxTest(unittest.TestCase): + + def test_exec_dmesg(self): + enable_networking = os.geteuid() == 0 + # Create the background sandbox + with sandbox.Sandbox(enable_networking=enable_networking) as sb: + # Execute dmesg in the gVisor sandbox. + stdout, _ = sb.exec("dmesg") + self.assertIn("Starting gVisor", stdout) + + def test_exec_timeout(self): + enable_networking = os.geteuid() == 0 + with sandbox.Sandbox(enable_networking=enable_networking) as sb: + with self.assertRaises(sandbox.SandboxError) as ctx: + sb.exec("sleep", "10", timeout=1) + self.assertIn("exec timed out", str(ctx.exception)) + + def test_exec_with_args(self): + enable_networking = os.geteuid() == 0 + with sandbox.Sandbox(enable_networking=enable_networking) as sb: + stdout, _ = sb.exec("echo", "hello", "sandbox") + self.assertEqual(stdout.strip(), "hello sandbox") + + def test_exec_invalid_command_or_args(self): + enable_networking = os.geteuid() == 0 + with sandbox.Sandbox(enable_networking=enable_networking) as sb: + with self.assertRaises(sandbox.SandboxError) as ctx: + sb.exec("nonexistent_command_xyz123") + self.assertIn("exec failed", str(ctx.exception)) + + with self.assertRaises(sandbox.SandboxError) as ctx: + sb.exec("ls", "--invalid-flag-xyz123") + self.assertIn("exec failed", str(ctx.exception)) + + + def test_sandbox_options(self): + enable_networking = os.geteuid() == 0 + with tempfile.TemporaryDirectory() as runtime_dir: + sandbox_id = "iwillbeasandbox" + with sandbox.Sandbox( + runtime_dir=runtime_dir, + sandbox_id=sandbox_id, + enable_networking=enable_networking, + ) as sb: + self.assertTrue(sb.bundle_dir.startswith(runtime_dir)) + self.assertEqual(sb.id, sandbox_id) + + def test_non_root_networking_error(self): + if os.geteuid() == 0: + self.skipTest("this test must be run as non-root") + + before_tmp = set(os.listdir(tempfile.gettempdir())) + with self.assertRaises(sandbox.SandboxError) as ctx: + sandbox.Sandbox(enable_networking=True) + after_tmp = set(os.listdir(tempfile.gettempdir())) + + self.assertIn( + "enabling networking requires running as root", str(ctx.exception) + ) + leaked = [f for f in after_tmp - before_tmp if f.startswith("gvisor-sandbox-")] + self.assertEqual(leaked, []) + + + def test_create_bundle(self): + # Test the internal _create_bundle method to verify config.json + for enable_networking in [False, True]: + with self.subTest(enable_networking=enable_networking): + sandbox_id = "test-sandbox" + + if enable_networking and os.geteuid() != 0: + continue + + with tempfile.TemporaryDirectory() as temp_dir: + try: + sb = sandbox.Sandbox( + runtime_dir=temp_dir, + sandbox_id=sandbox_id, + enable_networking=enable_networking, + ) + except sandbox.SandboxError as e: + self.fail(f"Failed to create sandbox: {e}") + + try: + bundle_dir = sb.bundle_dir + expected_bundle_dir = os.path.join(temp_dir, sandbox_id) + self.assertEqual(bundle_dir, expected_bundle_dir) + + config_path = os.path.join(bundle_dir, "config.json") + self.assertTrue(os.path.exists(config_path)) + + with open(config_path, "r") as f: + spec = json.load(f) + + self.assertEqual(spec.get("ociVersion"), "1.0.0") + self.assertEqual(spec.get("root", {}).get("path"), "rootfs") + self.assertTrue(spec.get("root", {}).get("readonly")) + + namespaces = spec.get("linux", {}).get("namespaces", []) + namespace_types = {ns.get("type") for ns in namespaces} + + expected_types = {"pid", "mount", "uts", "ipc"} + if os.geteuid() != 0: + expected_types.add("user") + if enable_networking: + expected_types.add("network") + + self.assertEqual(namespace_types, expected_types) + finally: + sb.close() + + self.assertFalse(os.path.exists(bundle_dir)) + + def test_runtime_dir_ownership_and_cleanup(self): + enable_networking = os.geteuid() == 0 + + # Auto-created runtime directory should be deleted on close. + sb = sandbox.Sandbox(enable_networking=enable_networking) + auto_runtime_dir = sb._runtime_dir + self.assertTrue(os.path.exists(auto_runtime_dir)) + sb.close() + self.assertFalse(os.path.exists(auto_runtime_dir)) + + # Custom runtime directory should not be deleted on close. + with tempfile.TemporaryDirectory() as custom_dir: + sb = sandbox.Sandbox( + runtime_dir=custom_dir, enable_networking=enable_networking + ) + sb.close() + self.assertTrue(os.path.exists(custom_dir)) + + def test_close_idempotent(self): + enable_networking = os.geteuid() == 0 + sb = sandbox.Sandbox(enable_networking=enable_networking) + sb.close() + # Repeating close() should be a safe no-op. + sb.close() + + +if __name__ == "__main__": + unittest.main()