Add support for custom network modes in Python SandboxExec.

sandboxexec/sandbox/python/gvisor/sandbox.py

@@ -39,6 +39,7 @@ def __init__(
       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 @@ def __init__(
       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,7 @@ def __init__(
     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:
         raise Error("enabling networking requires running as root")
 
       self._state_dir = os.path.join(self._runtime_dir, "state")
@@ -101,8 +115,12 @@ def __init__(
       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 +206,7 @@ def _create_bundle(self) -> str:
     ]
     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 = [

sandboxexec/sandbox/python/tests/sandbox_test.py

@@ -252,6 +252,82 @@ def side_effect(*args, **_kwargs):
       sandbox.Sandbox(enable_networking=False)
     self.assertIn("failed to create sandbox via subprocess", str(ctx.exception))
 
+  @mock.patch("subprocess.run")
+  def test_invalid_networking_mode(self, mock_run):  # 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_mode_namespaces(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)
+      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}
+      if net == "sandbox":
+        self.assertIn("network", namespace_types)
+      else:
+        self.assertNotIn("network", namespace_types)
+      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)
+  @mock.patch("subprocess.run")
+  def test_network_sandbox_nonroot_raises_error(
+      self, mock_run, mock_geteuid
+  ):  # pylint: disable=unused-argument
+    mock_run.return_value = mock.Mock(returncode=0)
+    with self.assertRaises(sandbox.Error) as ctx:
+      sandbox.Sandbox(network="sandbox")
+    self.assertIn(
+        "enabling networking requires running as root", str(ctx.exception)
+    )
+
+    sb = sandbox.Sandbox(network="none")
+    args = mock_run.call_args_list[0][0][0]
+    self.assertIn("--network=none", args)
+    sb.close()
+
   def test_find_runsc_not_found(self):
     old_runsc_path = os.environ.get("RUNSC_PATH")
     if "RUNSC_PATH" in os.environ: