Path resolution by kernel manager and providers

jupyter_client/manager.py

@@ -279,6 +279,11 @@ def client(self, **kwargs: t.Any) -> BlockingKernelClient:
     # Kernel management
     # --------------------------------------------------------------------------
 
+    def resolve_path(self, path: str) -> t.Optional[str]:
+        """Resolve path to given file."""
+        assert self.provisioner is not None
+        return self.provisioner.resolve_path(path)
+
     def update_env(self, *, env: t.Dict[str, str]) -> None:
         """
         Allow to update the environment of a kernel manager.

jupyter_client/provisioning/local_provisioner.py

@@ -3,6 +3,7 @@
 # Distributed under the terms of the Modified BSD License.
 import asyncio
 import os
+import pathlib
 import signal
 import sys
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
@@ -31,6 +32,7 @@ class LocalProvisioner(KernelProvisionerBase):  # type:ignore[misc]
     pgid = None
     ip = None
     ports_cached = False
+    cwd = None
 
     @property
     def has_process(self) -> bool:
@@ -206,6 +208,7 @@ async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
 
     async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnectionInfo:
         """Launch a kernel with a command."""
+
         scrubbed_kwargs = LocalProvisioner._scrub_kwargs(kwargs)
         self.process = launch_kernel(cmd, **scrubbed_kwargs)
         pgid = None
@@ -217,8 +220,18 @@ async def launch_kernel(self, cmd: List[str], **kwargs: Any) -> KernelConnection
 
         self.pid = self.process.pid
         self.pgid = pgid
+        self.cwd = kwargs.get("cwd", pathlib.Path.cwd())
         return self.connection_info
 
+    def resolve_path(self, path_str: str) -> Optional[str]:
+        """Resolve path to given file."""
+        path = pathlib.Path(path_str).expanduser()
+        if not path.is_absolute() and self.cwd:
+            path = (pathlib.Path(self.cwd) / path).resolve()
+        if path.exists():
+            return path.as_posix()
+        return None
+
     @staticmethod
     def _scrub_kwargs(kwargs: Dict[str, Any]) -> Dict[str, Any]:
         """Remove any keyword arguments that Popen does not tolerate."""

jupyter_client/provisioning/provisioner_base.py

@@ -219,6 +219,21 @@ def get_stable_start_time(self, recommended: float = 10.0) -> float:
         """
         return recommended
 
+    def resolve_path(self, path: str) -> Optional[str]:
+        """
+        Returns the path resolved relative to kernel working directory.
+
+        For example, path `my_code.py` for a kernel started in `/tmp/`
+        should result in `/tmp/my_code.py`, while path `~/test.py` for
+        a kernel started in `/home/my_user/` should resolve to the
+        (fully specified) `/home/my_user/test.py` path.
+
+        The provisioner may choose not to resolve any paths, or restrict
+        the resolution to paths local to the kernel working directory
+        to prevent path traversal and exposure of file system layout.
+        """
+        return None
+
     def _finalize_env(self, env: Dict[str, str]) -> None:
         """
         Ensures env is appropriate prior to launch.