Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Utilities for connecting to jupyter kernels
3The :class:`ConnectionFileMixin` class in this module encapsulates the logic
4related to writing and reading connections files.
5"""
7# Copyright (c) Jupyter Development Team.
8# Distributed under the terms of the Modified BSD License.
9from __future__ import annotations
11import errno
12import glob
13import json
14import os
15import socket
16import stat
17import tempfile
18import warnings
19from getpass import getpass
20from typing import TYPE_CHECKING, Any, cast
22import zmq
23from jupyter_core.paths import jupyter_data_dir, jupyter_runtime_dir, secure_write
24from traitlets import Bool, Bytes, CaselessStrEnum, Instance, Integer, Type, Unicode, observe
25from traitlets.config import LoggingConfigurable, SingletonConfigurable
26from typing_extensions import TypedDict
28from .localinterfaces import localhost
29from .utils import _filefind
31if TYPE_CHECKING:
32 from jupyter_client import BlockingKernelClient
34 from .session import Session
36# Define custom type for kernel connection info
39class KernelConnectionInfo(TypedDict, extra_items=str | bytes | int, total=False): # type: ignore[call-arg]
40 shell_port: int
41 iopub_port: int
42 stdin_port: int
43 control_port: int
44 hb_port: int
45 ip: str
46 key: str
47 transport: str
48 signature_scheme: str
49 kernel_name: str
50 session: Session
51 curve_publickey: str
52 curve_secretkey: str
55def write_connection_file(
56 fname: str | None = None,
57 shell_port: int = 0,
58 iopub_port: int = 0,
59 stdin_port: int = 0,
60 hb_port: int = 0,
61 control_port: int = 0,
62 ip: str = "",
63 key: bytes = b"",
64 transport: str = "tcp",
65 signature_scheme: str = "hmac-sha256",
66 kernel_name: str = "",
67 curve_publickey: bytes | None = None,
68 curve_secretkey: bytes | None = None,
69 **kwargs: Any,
70) -> tuple[str, KernelConnectionInfo]:
71 """Generates a JSON config file, including the selection of random ports.
73 Parameters
74 ----------
76 fname : unicode
77 The path to the file to write
79 shell_port : int, optional
80 The port to use for ROUTER (shell) channel.
82 iopub_port : int, optional
83 The port to use for the SUB channel.
85 stdin_port : int, optional
86 The port to use for the ROUTER (raw input) channel.
88 control_port : int, optional
89 The port to use for the ROUTER (control) channel.
91 hb_port : int, optional
92 The port to use for the heartbeat REP channel.
94 ip : str, optional
95 The ip address the kernel will bind to.
97 key : bytes, optional
98 The Session key used for message authentication.
100 signature_scheme : str, optional
101 The scheme used for message authentication.
102 This has the form 'digest-hash', where 'digest'
103 is the scheme used for digests, and 'hash' is the name of the hash function
104 used by the digest scheme.
105 Currently, 'hmac' is the only supported digest scheme,
106 and 'sha256' is the default hash function.
108 kernel_name : str, optional
109 The name of the kernel currently connected to.
111 curve_publickey : bytes, optional
112 CurveZMQ public key (Z85).
114 curve_secretkey : bytes, optional
115 CurveZMQ secret key (Z85).
116 """
117 if not ip:
118 ip = localhost()
119 # default to temporary connector file
120 if not fname:
121 fd, fname = tempfile.mkstemp(".json")
122 os.close(fd)
124 # Find open ports as necessary.
126 ports: list[int] = []
127 sockets: list[socket.socket] = []
128 ports_needed = (
129 int(shell_port <= 0)
130 + int(iopub_port <= 0)
131 + int(stdin_port <= 0)
132 + int(control_port <= 0)
133 + int(hb_port <= 0)
134 )
135 if transport == "tcp":
136 for _ in range(ports_needed):
137 sock = socket.socket()
138 # struct.pack('ii', (0,0)) is 8 null bytes
139 sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, b"\0" * 8)
140 sock.bind((ip, 0))
141 sockets.append(sock)
142 for sock in sockets:
143 port = sock.getsockname()[1]
144 sock.close()
145 ports.append(port)
146 else:
147 N = 1
148 for _ in range(ports_needed):
149 while os.path.exists(f"{ip}-{N!s}"):
150 N += 1
151 ports.append(N)
152 N += 1
153 if shell_port <= 0:
154 shell_port = ports.pop(0)
155 if iopub_port <= 0:
156 iopub_port = ports.pop(0)
157 if stdin_port <= 0:
158 stdin_port = ports.pop(0)
159 if control_port <= 0:
160 control_port = ports.pop(0)
161 if hb_port <= 0:
162 hb_port = ports.pop(0)
164 cfg: KernelConnectionInfo = {
165 "shell_port": shell_port,
166 "iopub_port": iopub_port,
167 "stdin_port": stdin_port,
168 "control_port": control_port,
169 "hb_port": hb_port,
170 }
171 cfg["ip"] = ip
172 cfg["key"] = key.decode()
173 cfg["transport"] = transport
174 cfg["signature_scheme"] = signature_scheme
175 cfg["kernel_name"] = kernel_name
176 if curve_publickey is not None:
177 cfg["curve_publickey"] = curve_publickey.decode("ascii")
178 if curve_secretkey is not None:
179 cfg["curve_secretkey"] = curve_secretkey.decode("ascii")
180 cfg.update(kwargs) # type: ignore[typeddict-item]
182 # Only ever write this file as user read/writeable
183 # This would otherwise introduce a vulnerability as a file has secrets
184 # which would let others execute arbitrary code as you
185 with secure_write(fname) as f:
186 f.write(json.dumps(cfg, indent=2))
188 if hasattr(stat, "S_ISVTX"):
189 # set the sticky bit on the parent directory of the file
190 # to ensure only owner can remove it
191 runtime_dir = os.path.dirname(fname)
192 if runtime_dir:
193 permissions = os.stat(runtime_dir).st_mode
194 new_permissions = permissions | stat.S_ISVTX
195 if new_permissions != permissions:
196 try:
197 os.chmod(runtime_dir, new_permissions)
198 except OSError as e:
199 if e.errno == errno.EPERM:
200 # suppress permission errors setting sticky bit on runtime_dir,
201 # which we may not own.
202 pass
203 return fname, cfg
206def find_connection_file(
207 filename: str = "kernel-*.json",
208 path: str | list[str] | None = None,
209 profile: str | None = None,
210) -> str:
211 """find a connection file, and return its absolute path.
213 The current working directory and optional search path
214 will be searched for the file if it is not given by absolute path.
216 If the argument does not match an existing file, it will be interpreted as a
217 fileglob, and the matching file in the profile's security dir with
218 the latest access time will be used.
220 Parameters
221 ----------
222 filename : str
223 The connection file or fileglob to search for.
224 path : str or list of strs[optional]
225 Paths in which to search for connection files.
227 Returns
228 -------
229 str : The absolute path of the connection file.
230 """
231 if profile is not None:
232 warnings.warn(
233 "Jupyter has no profiles. profile=%s has been ignored." % profile, stacklevel=2
234 )
235 if path is None:
236 path = [".", jupyter_runtime_dir()]
237 if isinstance(path, str):
238 path = [path]
240 try:
241 # first, try explicit name
242 return _filefind(filename, path)
243 except OSError:
244 pass
246 # not found by full name
248 if "*" in filename:
249 # given as a glob already
250 pat = filename
251 else:
252 # accept any substring match
253 pat = "*%s*" % filename
255 matches = []
256 for p in path:
257 matches.extend(glob.glob(os.path.join(p, pat)))
259 matches = [os.path.abspath(m) for m in matches]
260 if not matches:
261 msg = f"Could not find {filename!r} in {path!r}"
262 raise OSError(msg)
263 elif len(matches) == 1:
264 return matches[0]
265 else:
266 # get most recent match, by access time:
267 return sorted(matches, key=lambda f: os.stat(f).st_atime)[-1]
270def tunnel_to_kernel(
271 connection_info: str | KernelConnectionInfo,
272 sshserver: str,
273 sshkey: str | None = None,
274) -> tuple[Any, ...]:
275 """tunnel connections to a kernel via ssh
277 This will open five SSH tunnels from localhost on this machine to the
278 ports associated with the kernel. They can be either direct
279 localhost-localhost tunnels, or if an intermediate server is necessary,
280 the kernel must be listening on a public IP.
282 Parameters
283 ----------
284 connection_info : dict or str (path)
285 Either a connection dict, or the path to a JSON connection file
286 sshserver : str
287 The ssh sever to use to tunnel to the kernel. Can be a full
288 `user@server:port` string. ssh config aliases are respected.
289 sshkey : str [optional]
290 Path to file containing ssh key to use for authentication.
291 Only necessary if your ssh config does not already associate
292 a keyfile with the host.
294 Returns
295 -------
297 (shell, iopub, stdin, hb, control) : ints
298 The five ports on localhost that have been forwarded to the kernel.
299 """
300 from .ssh import tunnel
302 if isinstance(connection_info, str):
303 # it's a path, unpack it
304 with open(connection_info) as f:
305 connection_info = json.loads(f.read())
307 cf = cast(dict[str, Any], connection_info)
309 lports = tunnel.select_random_ports(5)
310 rports = (
311 cf["shell_port"],
312 cf["iopub_port"],
313 cf["stdin_port"],
314 cf["hb_port"],
315 cf["control_port"],
316 )
318 remote_ip = cf["ip"]
320 if tunnel.try_passwordless_ssh(sshserver, sshkey):
321 password: bool | str = False
322 else:
323 password = getpass("SSH Password for %s: " % sshserver)
325 for lp, rp in zip(lports, rports, strict=False):
326 tunnel.ssh_tunnel(lp, rp, sshserver, remote_ip, sshkey, password)
328 return tuple(lports)
331# -----------------------------------------------------------------------------
332# Mixin for classes that work with connection files
333# -----------------------------------------------------------------------------
335channel_socket_types = {
336 "hb": zmq.REQ,
337 "shell": zmq.DEALER,
338 "iopub": zmq.SUB,
339 "stdin": zmq.DEALER,
340 "control": zmq.DEALER,
341}
343port_names = ["%s_port" % channel for channel in ("shell", "stdin", "iopub", "hb", "control")]
346class ConnectionFileMixin(LoggingConfigurable):
347 """Mixin for configurable classes that work with connection files"""
349 data_dir: str | Unicode = Unicode()
351 def _data_dir_default(self) -> str:
352 return jupyter_data_dir()
354 # The addresses for the communication channels
355 connection_file = Unicode(
356 "",
357 config=True,
358 help="""JSON file in which to store connection info [default: kernel-<pid>.json]
360 This file will contain the IP, ports, and authentication key needed to connect
361 clients to this kernel. By default, this file will be created in the security dir
362 of the current profile, but can be specified by absolute path.
363 """,
364 )
365 _connection_file_written = Bool(False)
367 transport = CaselessStrEnum(["tcp", "ipc"], default_value="tcp", config=True)
368 kernel_name: str | Unicode = Unicode()
370 context = Instance(zmq.Context)
372 ip = Unicode(
373 config=True,
374 help="""Set the kernel\'s IP address [default localhost].
375 If the IP address is something other than localhost, then
376 Consoles on other machines will be able to connect
377 to the Kernel, so be careful!""",
378 )
380 def _ip_default(self) -> str:
381 if self.transport == "ipc":
382 if self.connection_file:
383 return os.path.splitext(self.connection_file)[0] + "-ipc"
384 else:
385 return "kernel-ipc"
386 else:
387 return localhost()
389 @observe("ip")
390 def _ip_changed(self, change: Any) -> None:
391 if change["new"] == "*":
392 self.ip = "0.0.0.0" # noqa
394 # protected traits
396 hb_port = Integer(0, config=True, help="set the heartbeat port [default: random]")
397 shell_port = Integer(0, config=True, help="set the shell (ROUTER) port [default: random]")
398 iopub_port = Integer(0, config=True, help="set the iopub (PUB) port [default: random]")
399 stdin_port = Integer(0, config=True, help="set the stdin (ROUTER) port [default: random]")
400 control_port = Integer(0, config=True, help="set the control (ROUTER) port [default: random]")
402 # Optional CurveZMQ keys loaded from the connection file (Z85-encoded bytes).
403 # None when the kernel was not started with CurveZMQ enabled.
404 curve_publickey: Bytes | None = Bytes(allow_none=True, default_value=None)
405 curve_secretkey: Bytes | None = Bytes(allow_none=True, default_value=None)
407 # names of the ports with random assignment
408 _random_port_names: list[str] | None = None
410 @property
411 def ports(self) -> list[int]:
412 return [getattr(self, name) for name in port_names]
414 # The Session to use for communication with the kernel.
415 session = Instance("jupyter_client.session.Session")
417 def _session_default(self) -> Session:
418 from .session import Session
420 return Session(parent=self)
422 # --------------------------------------------------------------------------
423 # Connection and ipc file management
424 # --------------------------------------------------------------------------
426 def get_connection_info(self, session: bool = False) -> KernelConnectionInfo:
427 """Return the connection info as a dict
429 Parameters
430 ----------
431 session : bool [default: False]
432 If True, return our session object will be included in the connection info.
433 If False (default), the configuration parameters of our session object will be included,
434 rather than the session object itself.
436 Returns
437 -------
438 connect_info : dict
439 dictionary of connection information.
440 """
441 info: KernelConnectionInfo = {
442 "transport": self.transport,
443 "ip": self.ip,
444 "shell_port": self.shell_port,
445 "iopub_port": self.iopub_port,
446 "stdin_port": self.stdin_port,
447 "hb_port": self.hb_port,
448 "control_port": self.control_port,
449 }
450 if session:
451 # add *clone* of my session,
452 # so that state such as digest_history is not shared.
453 info["session"] = self.session.clone()
454 else:
455 # add session info
456 info.update(
457 {
458 "signature_scheme": self.session.signature_scheme,
459 "key": self.session.key,
460 }
461 )
462 if self.curve_publickey is not None and self.curve_secretkey is not None:
463 info["curve_publickey"] = self.curve_publickey.decode()
464 info["curve_secretkey"] = self.curve_secretkey.decode()
465 return info
467 # factory for blocking clients
468 blocking_class = Type(klass=object, default_value="jupyter_client.BlockingKernelClient")
470 def blocking_client(self) -> BlockingKernelClient:
471 """Make a blocking client connected to my kernel"""
472 info = self.get_connection_info()
473 bc = self.blocking_class(parent=self) # type:ignore[operator]
474 bc.load_connection_info(info)
475 return bc
477 def cleanup_connection_file(self) -> None:
478 """Cleanup connection file *if we wrote it*
480 Will not raise if the connection file was already removed somehow.
481 """
482 if self._connection_file_written:
483 # cleanup connection files on full shutdown of kernel we started
484 self._connection_file_written = False
485 try:
486 os.remove(self.connection_file)
487 except (OSError, AttributeError):
488 pass
490 def cleanup_ipc_files(self) -> None:
491 """Cleanup ipc files if we wrote them."""
492 if self.transport != "ipc":
493 return
494 for port in self.ports:
495 ipcfile = "%s-%i" % (self.ip, port)
496 try:
497 os.remove(ipcfile)
498 except OSError:
499 pass
501 def _record_random_port_names(self) -> None:
502 """Records which of the ports are randomly assigned.
504 Records on first invocation, if the transport is tcp.
505 Does nothing on later invocations."""
507 if self.transport != "tcp":
508 return
509 if self._random_port_names is not None:
510 return
512 self._random_port_names = []
513 for name in port_names:
514 if getattr(self, name) <= 0:
515 self._random_port_names.append(name)
517 def cleanup_random_ports(self) -> None:
518 """Forgets randomly assigned port numbers and cleans up the connection file.
520 Does nothing if no port numbers have been randomly assigned.
521 In particular, does nothing unless the transport is tcp.
522 """
524 if not self._random_port_names:
525 return
527 for name in self._random_port_names:
528 setattr(self, name, 0)
530 self.cleanup_connection_file()
532 def write_connection_file(self, **kwargs: Any) -> None:
533 """Write connection info to JSON dict in self.connection_file."""
534 if self._connection_file_written and os.path.exists(self.connection_file):
535 return
537 self.connection_file, cfg = write_connection_file(
538 self.connection_file,
539 transport=self.transport,
540 ip=self.ip,
541 key=self.session.key,
542 stdin_port=self.stdin_port,
543 iopub_port=self.iopub_port,
544 shell_port=self.shell_port,
545 hb_port=self.hb_port,
546 control_port=self.control_port,
547 signature_scheme=self.session.signature_scheme,
548 kernel_name=self.kernel_name,
549 curve_publickey=self.curve_publickey,
550 curve_secretkey=self.curve_secretkey,
551 **kwargs,
552 )
553 # write_connection_file also sets default ports:
554 self._record_random_port_names()
555 for name in port_names:
556 setattr(self, name, cast(int, cfg.get(name)))
558 self._connection_file_written = True
560 def load_connection_file(self, connection_file: str | None = None) -> None:
561 """Load connection info from JSON dict in self.connection_file.
563 Parameters
564 ----------
565 connection_file: unicode, optional
566 Path to connection file to load.
567 If unspecified, use self.connection_file
568 """
569 if connection_file is None:
570 connection_file = self.connection_file
571 self.log.debug("Loading connection file %s", connection_file)
572 with open(connection_file) as f:
573 info = json.load(f)
574 self.load_connection_info(info)
576 def load_connection_info(self, info: KernelConnectionInfo) -> None:
577 """Load connection info from a dict containing connection info.
579 Typically this data comes from a connection file
580 and is called by load_connection_file.
582 Parameters
583 ----------
584 info: dict
585 Dictionary containing connection_info.
586 See the connection_file spec for details.
587 """
588 self.transport = info.get("transport", self.transport)
589 self.ip = info.get("ip", self._ip_default())
591 self._record_random_port_names()
592 for name in port_names:
593 if getattr(self, name) == 0 and name in info:
594 # not overridden by config or cl_args
595 setattr(self, name, cast(int, info.get(name)))
597 if "key" in info:
598 key = info["key"]
599 key_bytes = key if isinstance(key, bytes) else key.encode() # type: ignore[redundant-expr,unreachable]
600 self.session.key = key_bytes
601 if "signature_scheme" in info:
602 self.session.signature_scheme = info["signature_scheme"]
603 if "curve_publickey" in info and "curve_secretkey" in info:
604 pub = info["curve_publickey"]
605 sec = info["curve_secretkey"]
606 self.curve_publickey = pub.encode() if isinstance(pub, str) else pub # type: ignore[redundant-expr]
607 self.curve_secretkey = sec.encode() if isinstance(sec, str) else sec # type: ignore[redundant-expr]
609 def _reconcile_connection_info(self, info: KernelConnectionInfo) -> None:
610 """Reconciles the connection information returned from the Provisioner.
612 Because some provisioners (like derivations of LocalProvisioner) may have already
613 written the connection file, this method needs to ensure that, if the connection
614 file exists, its contents match that of what was returned by the provisioner. If
615 the file does exist and its contents do not match, the file will be replaced with
616 the provisioner information (which is considered the truth).
618 If the file does not exist, the connection information in 'info' is loaded into the
619 KernelManager and written to the file.
620 """
621 # Prevent over-writing a file that has already been written with the same
622 # info. This is to prevent a race condition where the process has
623 # already been launched but has not yet read the connection file - as is
624 # the case with LocalProvisioners.
625 file_exists: bool = False
626 if os.path.exists(self.connection_file):
627 with open(self.connection_file) as f:
628 file_info = json.load(f)
629 # Prior to the following comparison, we need to adjust the value of "key" to
630 # be bytes, otherwise the comparison below will fail.
631 file_info["key"] = file_info["key"].encode()
632 if not self._equal_connections(info, file_info):
633 os.remove(self.connection_file) # Contents mismatch - remove the file
634 self._connection_file_written = False
635 else:
636 file_exists = True
638 if not file_exists:
639 # Load the connection info and write out file, clearing existing
640 # port-based attributes so they will be reloaded
641 for name in port_names:
642 setattr(self, name, 0)
643 self.load_connection_info(info)
644 self.write_connection_file()
646 # Ensure what is in KernelManager is what we expect.
647 km_info = self.get_connection_info()
648 if not self._equal_connections(info, km_info):
649 msg = (
650 "KernelManager's connection information already exists and does not match "
651 "the expected values returned from provisioner!"
652 )
653 raise ValueError(msg)
655 @staticmethod
656 def _equal_connections(conn1: KernelConnectionInfo, conn2: KernelConnectionInfo) -> bool:
657 """Compares pertinent keys of connection info data. Returns True if equivalent, False otherwise."""
659 pertinent_keys = [
660 "key",
661 "curve_publickey",
662 "curve_secretkey",
663 "ip",
664 "stdin_port",
665 "iopub_port",
666 "shell_port",
667 "control_port",
668 "hb_port",
669 "transport",
670 "signature_scheme",
671 ]
673 return all(conn1.get(key) == conn2.get(key) for key in pertinent_keys)
675 # --------------------------------------------------------------------------
676 # Creating connected sockets
677 # --------------------------------------------------------------------------
679 def _make_url(self, channel: str) -> str:
680 """Make a ZeroMQ URL for a given channel."""
681 transport = self.transport
682 ip = self.ip
683 port = getattr(self, "%s_port" % channel)
685 if transport == "tcp":
686 return "tcp://%s:%i" % (ip, port)
687 else:
688 return f"{transport}://{ip}-{port}"
690 def _create_connected_socket(
691 self, channel: str, identity: bytes | None = None
692 ) -> zmq.sugar.socket.Socket:
693 """Create a zmq Socket and connect it to the kernel."""
694 url = self._make_url(channel)
695 socket_type = channel_socket_types[channel]
696 self.log.debug("Connecting to: %s", url)
697 sock = self.context.socket(socket_type)
698 # set linger to 1s to prevent hangs at exit
699 sock.linger = 1000
700 if identity:
701 sock.identity = identity
702 if self.curve_publickey is not None:
703 # The connection file already carries this keypair, so reusing it
704 # avoids introducing an additional key-distribution mechanism here.
705 # curve_serverkey authenticates the server; the keypair configures
706 # encrypted communication for the client socket.
707 sock.curve_secretkey = self.curve_secretkey
708 sock.curve_publickey = self.curve_publickey
709 sock.curve_serverkey = self.curve_publickey
710 sock.connect(url)
711 return sock
713 def connect_iopub(self, identity: bytes | None = None) -> zmq.sugar.socket.Socket:
714 """return zmq Socket connected to the IOPub channel"""
715 sock = self._create_connected_socket("iopub", identity=identity)
716 sock.setsockopt(zmq.SUBSCRIBE, b"")
717 return sock
719 def connect_shell(self, identity: bytes | None = None) -> zmq.sugar.socket.Socket:
720 """return zmq Socket connected to the Shell channel"""
721 return self._create_connected_socket("shell", identity=identity)
723 def connect_stdin(self, identity: bytes | None = None) -> zmq.sugar.socket.Socket:
724 """return zmq Socket connected to the StdIn channel"""
725 return self._create_connected_socket("stdin", identity=identity)
727 def connect_hb(self, identity: bytes | None = None) -> zmq.sugar.socket.Socket:
728 """return zmq Socket connected to the Heartbeat channel"""
729 return self._create_connected_socket("hb", identity=identity)
731 def connect_control(self, identity: bytes | None = None) -> zmq.sugar.socket.Socket:
732 """return zmq Socket connected to the Control channel"""
733 return self._create_connected_socket("control", identity=identity)
736class LocalPortCache(SingletonConfigurable):
737 """
738 Used to keep track of local ports in order to prevent race conditions that
739 can occur between port acquisition and usage by the kernel. All locally-
740 provisioned kernels should use this mechanism to limit the possibility of
741 race conditions. Note that this does not preclude other applications from
742 acquiring a cached but unused port, thereby re-introducing the issue this
743 class is attempting to resolve (minimize).
744 See: https://github.com/jupyter/jupyter_client/issues/487
745 """
747 def __init__(self, **kwargs: Any) -> None:
748 super().__init__(**kwargs)
749 self.currently_used_ports: set[int] = set()
751 def find_available_port(self, ip: str) -> int:
752 while True:
753 tmp_sock = socket.socket()
754 tmp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, b"\0" * 8)
755 tmp_sock.bind((ip, 0))
756 port = tmp_sock.getsockname()[1]
757 tmp_sock.close()
759 # This is a workaround for https://github.com/jupyter/jupyter_client/issues/487
760 # We prevent two kernels to have the same ports.
761 if port not in self.currently_used_ports:
762 self.currently_used_ports.add(port)
763 return port
765 def return_port(self, port: int) -> None:
766 if port in self.currently_used_ports: # Tolerate uncached ports
767 self.currently_used_ports.remove(port)
770__all__ = [
771 "KernelConnectionInfo",
772 "LocalPortCache",
773 "find_connection_file",
774 "tunnel_to_kernel",
775 "write_connection_file",
776]