Add a toggle to enable curve encryption for all kernels that support it

jupyter_server/services/kernels/kernelmanager.py

@@ -67,6 +67,16 @@ def _default_kernel_manager_class(self):
 
     kernel_argv = List(Unicode())
 
+    transport_encryption = Bool(
+        False,
+        config=True,
+        help=(
+            "Enable transport encryption using manager-provisioned CurveZMQ keys for all managed kernels. "
+            "When True, the kernel manager/provisioner launch path is instructed "
+            "to provision per-kernel Curve credentials."
+        ),
+    )
+
     root_dir = Unicode(config=True)
 
     _kernel_connections = Dict()
@@ -204,6 +214,13 @@ def cwd_for_path(self, path, **kwargs):
             os_path = os.path.dirname(os_path)
         return os_path
 
+    def _kernel_start_kwargs(self, **kwargs: t.Any) -> dict[str, t.Any]:
+        """Build kernel launch kwargs with server-level policy applied."""
+        launch_kwargs = dict(kwargs)
+        if self.transport_encryption:
+            launch_kwargs["transport_encryption"] = True
+        return launch_kwargs
+
     async def _remove_kernel_when_ready(self, kernel_id, kernel_awaitable):
         """Remove a kernel when it is ready."""
         await super()._remove_kernel_when_ready(kernel_id, kernel_awaitable)
@@ -213,7 +230,7 @@ async def _remove_kernel_when_ready(self, kernel_id, kernel_awaitable):
     # TODO: DEC 2022: Revise the type-ignore once the signatures have been changed upstream
     # https://github.com/jupyter/jupyter_client/pull/905
     async def _async_start_kernel(  # type:ignore[override]
-        self, *, kernel_id: str | None = None, path: ApiPath | None = None, **kwargs: str
+        self, *, kernel_id: str | None = None, path: ApiPath | None = None, **kwargs: t.Any
     ) -> str:
         """Start a kernel for a session and return its kernel_id.
 
@@ -231,6 +248,7 @@ async def _async_start_kernel(  # type:ignore[override]
             an existing kernel is returned, but it may be checked in the future.
         """
         if kernel_id is None or kernel_id not in self:
+            kwargs = self._kernel_start_kwargs(**kwargs)
             if path is not None:
                 kwargs["cwd"] = self.cwd_for_path(path, env=kwargs.get("env", {}))
             if kernel_id is not None:

tests/services/kernels/test_config.py

@@ -1,7 +1,10 @@
 import pytest
 from traitlets.config import Config
 
-from jupyter_server.services.kernels.kernelmanager import AsyncMappingKernelManager
+from jupyter_server.services.kernels.kernelmanager import (
+    AsyncMappingKernelManager,
+    MappingKernelManager,
+)
 
 
 @pytest.fixture
@@ -29,3 +32,12 @@ def test_not_server_kernel_manager(jp_configurable_serverapp):
     ]
     with pytest.warns(FutureWarning, match="is not a subclass of 'ServerKernelManager'"):
         jp_configurable_serverapp(argv=argv)
+
+
+def test_kernel_start_kwargs_transport_encryption_sets_flag():
+    km = MappingKernelManager(transport_encryption=True)
+
+    launch_kwargs = km._kernel_start_kwargs(env={"EXISTING": "1"})
+
+    assert launch_kwargs["transport_encryption"] is True
+    assert launch_kwargs["env"] == {"EXISTING": "1"}