Make the client own key generation

Author: krassowski

ipykernel/kernelapp.py

@@ -188,19 +188,6 @@ def abs_connection_file(self):
     """,
     ).tag(config=True)
 
-    enable_curve = Bool(
-        bool(int(os.environ.get("JUPYTER_ENABLE_CURVE", "0"))),
-        help="Enable CurveZMQ transport encryption and authentication. "
-        "When True, a keypair is generated at startup and stored in the "
-        "connection file so that clients can authenticate and encrypt "
-        "all ZMQ channels.",
-    ).tag(config=True)
-
-    # Internal CurveZMQ keypair (Z85-encoded bytes); populated in init_sockets
-    # when enable_curve is True.
-    _curve_publickey: bytes | None = None
-    _curve_secretkey: bytes | None = None
-
     # polling
     parent_handle = Integer(
         int(os.environ.get("JPY_PARENT_PID") or 0),
@@ -227,12 +214,12 @@ def excepthook(self, etype, evalue, tb):
     def _apply_curve_server_options(self, socket: zmq.Socket[t.Any]) -> None:
         """Set CurveZMQ server-side options on *socket* before it is bound.
 
-        This is a no-op when enable_curve is False or keys have not been
-        generated yet, so it is safe to call unconditionally.
+        This is a no-op when Curve keys are not available yet, so it is safe
+        to call unconditionally.
         """
-        if self.enable_curve and self._curve_secretkey is not None:
-            socket.curve_secretkey = self._curve_secretkey
-            socket.curve_publickey = self._curve_publickey
+        if self.curve_secretkey is not None:
+            socket.curve_secretkey = self.curve_secretkey
+            socket.curve_publickey = self.curve_publickey
             socket.curve_server = True
 
     def init_poller(self):
@@ -298,10 +285,9 @@ def write_connection_file(self, **kwargs: Any) -> None:
             iopub_port=self.iopub_port,
             control_port=self.control_port,
         )
-        if self.enable_curve and self._curve_publickey is not None:
-            # write_connection_file() in jupyter-client handles JSON-safe key serialization
-            connection_info["curve_publickey"] = self._curve_publickey
-            connection_info["curve_secretkey"] = self._curve_secretkey
+        if self.curve_publickey is not None:
+            connection_info["curve_publickey"] = self.curve_publickey
+            connection_info["curve_secretkey"] = self.curve_secretkey
         if Path(cf).exists():
             # If the file exists, merge our info into it. For example, if the
             # original file had port number 0, we update with the actual port
@@ -356,16 +342,15 @@ def init_sockets(self):
         self.context = context = zmq.Context()
         atexit.register(self.close)
 
-        if self.enable_curve:
-            self._curve_publickey, self._curve_secretkey = zmq.curve_keypair()
-            self.log.debug("CurveZMQ enabled; generated server keypair")
+        if self.curve_secretkey is not None:
+            self.log.debug("Detected CurveZMQ secret key; using transport encryption")
         elif self.transport == "tcp":
             self.log.warning(
                 "Kernel is running over TCP without encryption."
                 " All communication (including code and outputs) is sent in plain text"
                 " and is susceptible to eavesdropping."
-                " Use IPC transport or set IPKernelApp.enable_curve=True to enable"
-                " CurveZMQ encryption."
+                " Use IPC transport or launch with kernel manager-provisioned"
+                " CurveZMQ keys to enable transport encryption."
             )
 
         self.shell_socket = context.socket(zmq.ROUTER)
@@ -439,8 +424,8 @@ def init_heartbeat(self):
         self.heartbeat = Heartbeat(
             hb_ctx,
             (self.transport, self.ip, self.hb_port),
-            curve_publickey=self._curve_publickey if self.enable_curve else None,
-            curve_secretkey=self._curve_secretkey if self.enable_curve else None,
+            curve_publickey=self.curve_publickey,
+            curve_secretkey=self.curve_secretkey,
         )
         self.hb_port = self.heartbeat.port
         self.log.debug("Heartbeat REP Channel on port: %i", self.hb_port)

tests/test_curve.py

@@ -6,6 +6,7 @@
 
 import pytest
 import zmq
+from jupyter_client import KernelManager
 
 from ipykernel.kernelapp import IPKernelApp
 
@@ -28,12 +29,6 @@ def curve_enabled_kernel_app(tmp_path):
         app.close()
 
 
-def test_curve_disabled_by_default():
-    """CurveZMQ must be off by default so existing kernels are unaffected."""
-    app = IPKernelApp()
-    assert app.enable_curve is False
-
-
 def test_connection_file_no_curve_keys_by_default(curve_disabled_kernel_app):
     """Connection file must not contain curve keys when Curve is disabled."""
     app, connection_file_path = curve_disabled_kernel_app
@@ -65,14 +60,14 @@ def test_curve_connection_file_has_keys(curve_enabled_kernel_app):
 
 
 def test_curve_keys_are_stable_per_startup(curve_enabled_kernel_app):
-    """Keys generated at startup stay the same throughout the process lifetime."""
+    """Provisioned keys stay unchanged throughout the kernel process lifetime."""
     app, _connection_file_path = curve_enabled_kernel_app
     app.init_sockets()
-    pub1 = app._curve_publickey
+    pub1 = app.curve_publickey
     # Writing the file twice should not regenerate keys.
     app.init_heartbeat()
     app.write_connection_file()
-    assert app._curve_publickey == pub1
+    assert app.curve_publickey == pub1
 
 
 def test_curve_socket_server_options(curve_enabled_kernel_app):
@@ -134,7 +129,7 @@ def test_curve_authenticated_socket_can_communicate(curve_enabled_kernel_app):
     app.init_sockets()
 
     endpoint = f"tcp://{app.ip}:{app.shell_port}"
-    server_public = app._curve_publickey
+    server_public = app.curve_publickey
 
     ctx = zmq.Context()
     auth_client = ctx.socket(zmq.DEALER)
@@ -161,9 +156,33 @@ def test_curve_authenticated_socket_can_communicate(curve_enabled_kernel_app):
         ctx.term()
 
 
-def _make_app(tmp_path, **kwargs):
+def test_manager_provisioned_curve_keys_are_used(curve_enabled_kernel_app):
+    """Kernel uses manager-provisioned Curve keys exactly as provided."""
+    app, _connection_file_path = curve_enabled_kernel_app
+    try:
+        with open(_connection_file_path) as f:
+            info = json.load(f)
+
+        app.init_sockets()
+
+        assert app.curve_publickey == info["curve_publickey"].encode()
+        assert app.curve_secretkey == info["curve_secretkey"].encode()
+        assert app.shell_socket.curve_server
+        assert app.stdin_socket.curve_server
+        assert app.control_socket.curve_server
+    finally:
+        app.close()
+
+
+def _make_app(tmp_path, *, enable_curve=False, **kwargs):
     """Return a minimal IPKernelApp rooted in temporary directory *tmp_path*."""
     connection_file_path = str(tmp_path / "kernel.json")
+    if enable_curve:
+        # Populate the Curve keys into the connection file
+        km = KernelManager(connection_file=connection_file_path)
+        km.transport_encryption = True
+        km.pre_start_kernel()
+
     app = IPKernelApp(connection_file=connection_file_path, **kwargs)
     # Replicate the subset of initialize() that sets up connection info
     # without importing IPython shell machinery.

tests/test_kernelapp.py

@@ -5,6 +5,7 @@
 from unittest.mock import patch
 
 import pytest
+from jupyter_client import KernelManager
 from jupyter_core.paths import secure_write
 from traitlets.config.loader import Config
 
@@ -131,25 +132,32 @@ def test_trio_loop():
     app.close()
 
 
-def test_init_sockets_curve_enabled_logs_debug():
-    app = IPKernelApp(enable_curve=True)
+def test_init_sockets_curve_enabled_logs_debug(tmp_path):
+    connection_file = str(tmp_path / "kernel.json")
+    km = KernelManager(connection_file=connection_file)
+    km.transport_encryption = True
+    km.pre_start_kernel()
+
+    app = IPKernelApp(connection_file=connection_file)
+    super(IPKernelApp, app).initialize(argv=[""])
+    app.init_connection_file()
     with patch.object(app.log, "debug") as mock_debug:
         app.init_sockets()
     app.cleanup_connection_file()
     app.close()
     messages = [str(call) for call in mock_debug.call_args_list]
-    assert any("CurveZMQ enabled" in m for m in messages), (
-        "Expected a debug log mentioning CurveZMQ when enable_curve=True"
+    assert any("Detected CurveZMQ secret key; using transport encryption" in m for m in messages), (
+        "Expected a debug log mentioning CurveZMQ when keys are provided"
     )
 
 
 def test_init_sockets_tcp_without_curve_logs_warning():
-    app = IPKernelApp(transport="tcp", enable_curve=False)
+    app = IPKernelApp(transport="tcp")
     with patch.object(app.log, "warning") as mock_warning:
         app.init_sockets()
     app.cleanup_connection_file()
     app.close()
     messages = [str(call) for call in mock_warning.call_args_list]
     assert any("Kernel is running over TCP without encryption" in m for m in messages), (
-        "Expected a warning about missing encryption when transport=tcp and enable_curve=False"
+        "Expected a warning about missing encryption when transport=tcp without curve keys"
     )