mirror of
https://github.com/hanzoai/visor.git
synced 2026-09-24 17:04:14 +00:00
Add support for custom network modes in Python SandboxExec.
PiperOrigin-RevId: 960325662
This commit is contained in:
@@ -39,6 +39,7 @@ class Sandbox:
|
||||
runtime_dir: Optional[str] = None,
|
||||
sandbox_id: Optional[str] = None,
|
||||
enable_networking: bool = True,
|
||||
network: Optional[str] = None,
|
||||
):
|
||||
"""Initializes and starts a new sandbox.
|
||||
|
||||
@@ -48,11 +49,24 @@ class Sandbox:
|
||||
sandbox_id: Specific sandbox ID. If not set, a unique ID is generated
|
||||
automatically.
|
||||
enable_networking: Whether networking is enabled inside the sandbox.
|
||||
network: The networking mode for runsc (e.g. "none", "sandbox", "host").
|
||||
Specifying this overrides enable_networking.
|
||||
|
||||
Raises:
|
||||
Error: If sandbox creation fails.
|
||||
ValueError: If an invalid network mode is provided.
|
||||
"""
|
||||
if network is not None and network not in ("none", "sandbox", "host"):
|
||||
raise ValueError(
|
||||
f"Invalid network mode '{network}'. Valid options are 'none',"
|
||||
" 'sandbox', 'host', or None."
|
||||
)
|
||||
|
||||
self._enable_networking = enable_networking
|
||||
self._network = network
|
||||
self._is_network_enabled = (
|
||||
network != "none" if network is not None else enable_networking
|
||||
)
|
||||
self._runtime_dir = ""
|
||||
self._owns_runtime_dir = False
|
||||
self._id = ""
|
||||
@@ -74,7 +88,11 @@ class Sandbox:
|
||||
self._id = sandbox_id or self._generate_id()
|
||||
|
||||
try:
|
||||
if os.geteuid() != 0 and self._enable_networking:
|
||||
if (
|
||||
os.geteuid() != 0
|
||||
and self._is_network_enabled
|
||||
and self._network != "host"
|
||||
):
|
||||
raise Error("enabling networking requires running as root")
|
||||
|
||||
self._state_dir = os.path.join(self._runtime_dir, "state")
|
||||
@@ -101,8 +119,12 @@ class Sandbox:
|
||||
args = ["--root", self._state_dir]
|
||||
if os.geteuid() != 0:
|
||||
args.append("--ignore-cgroups")
|
||||
if not self._enable_networking:
|
||||
|
||||
if self._network is not None:
|
||||
args.append(f"--network={self._network}")
|
||||
elif 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
|
||||
@@ -188,7 +210,7 @@ class Sandbox:
|
||||
]
|
||||
if os.geteuid() != 0:
|
||||
namespaces.append({"type": "user"})
|
||||
if self._enable_networking:
|
||||
if self._is_network_enabled and self._network != "host":
|
||||
namespaces.append({"type": "network"})
|
||||
|
||||
mounts = [
|
||||
|
||||
@@ -252,6 +252,80 @@ class SandboxTest(unittest.TestCase):
|
||||
sandbox.Sandbox(enable_networking=False)
|
||||
self.assertIn("failed to create sandbox via subprocess", str(ctx.exception))
|
||||
|
||||
def test_invalid_networking_mode(self): # pylint: disable=unused-argument
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
sandbox.Sandbox(network="invalid-net")
|
||||
self.assertIn(
|
||||
"Invalid network mode 'invalid-net'. Valid options are 'none',"
|
||||
" 'sandbox', 'host', or None.",
|
||||
str(ctx.exception),
|
||||
)
|
||||
|
||||
@mock.patch("os.geteuid", return_value=0)
|
||||
@mock.patch("subprocess.run")
|
||||
def test_network_modes(self, mock_run, mock_geteuid): # pylint: disable=unused-argument
|
||||
mock_run.return_value = mock.Mock(returncode=0)
|
||||
for net in ["none", "sandbox", "host"]:
|
||||
mock_run.reset_mock()
|
||||
sb = sandbox.Sandbox(network=net)
|
||||
args = mock_run.call_args_list[0][0][0]
|
||||
if net != "none":
|
||||
self.assertIn(f"--network={net}", args)
|
||||
else:
|
||||
self.assertIn("--network=none", args)
|
||||
sb.close()
|
||||
|
||||
@mock.patch("os.geteuid", return_value=0)
|
||||
@mock.patch("subprocess.run")
|
||||
def test_network_none_overrides_enable_networking_true(
|
||||
self, mock_run, mock_geteuid
|
||||
): # pylint: disable=unused-argument
|
||||
mock_run.return_value = mock.Mock(returncode=0)
|
||||
sb = sandbox.Sandbox(enable_networking=True, network="none")
|
||||
args = mock_run.call_args_list[0][0][0]
|
||||
self.assertIn("--network=none", args)
|
||||
config_path = os.path.join(sb.bundle_dir, "config.json")
|
||||
with open(config_path, "r") as f:
|
||||
spec = json.load(f)
|
||||
namespaces = spec.get("linux", {}).get("namespaces", [])
|
||||
namespace_types = {ns.get("type") for ns in namespaces}
|
||||
self.assertNotIn("network", namespace_types)
|
||||
sb.close()
|
||||
|
||||
@mock.patch("os.geteuid", return_value=1000)
|
||||
def test_network_sandbox_nonroot_raises_error(
|
||||
self, mock_geteuid
|
||||
): # pylint: disable=unused-argument
|
||||
with self.assertRaises(sandbox.Error) as ctx:
|
||||
sandbox.Sandbox(network="sandbox")
|
||||
self.assertIn(
|
||||
"enabling networking requires running as root", str(ctx.exception)
|
||||
)
|
||||
|
||||
def test_network_mode_none_real_sandbox(self):
|
||||
"""Verifies starting a real sandbox with network='none' without mocks."""
|
||||
with sandbox.Sandbox(network="none") as sb:
|
||||
stdout, _ = sb.exec("echo", "hello network none")
|
||||
self.assertEqual(stdout.strip(), "hello network none")
|
||||
config_path = os.path.join(sb.bundle_dir, "config.json")
|
||||
with open(config_path, "r") as f:
|
||||
spec = json.load(f)
|
||||
namespaces = spec.get("linux", {}).get("namespaces", [])
|
||||
namespace_types = {ns.get("type") for ns in namespaces}
|
||||
self.assertNotIn("network", namespace_types)
|
||||
|
||||
def test_network_mode_host_real_sandbox(self):
|
||||
"""Verifies starting a real sandbox with network='host' without mocks."""
|
||||
with sandbox.Sandbox(network="host") as sb:
|
||||
stdout, _ = sb.exec("echo", "hello network host")
|
||||
self.assertEqual(stdout.strip(), "hello network host")
|
||||
config_path = os.path.join(sb.bundle_dir, "config.json")
|
||||
with open(config_path, "r") as f:
|
||||
spec = json.load(f)
|
||||
namespaces = spec.get("linux", {}).get("namespaces", [])
|
||||
namespace_types = {ns.get("type") for ns in namespaces}
|
||||
self.assertNotIn("network", namespace_types)
|
||||
|
||||
def test_find_runsc_not_found(self):
|
||||
old_runsc_path = os.environ.get("RUNSC_PATH")
|
||||
if "RUNSC_PATH" in os.environ:
|
||||
|
||||
Reference in New Issue
Block a user