Path resolution by kernel manager and providers (#1005)

* Path resolution by kernel manager and providers

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Cleanup

* Do not resolve for non-existent files

* Fix spurious `async`

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update resolve_path return type to str | None

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: krassowski

jupyter_client/manager.py

@@ -280,6 +280,11 @@ def client(self, **kwargs: t.Any) -> BlockingKernelClient:
     # Kernel management
     # --------------------------------------------------------------------------
 
+    def resolve_path(self, path: str) -> str | None:
+        """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

@@ -4,6 +4,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
@@ -32,6 +33,7 @@ class LocalProvisioner(KernelProvisionerBase):
     pgid = None
     ip = None
     ports_cached = False
+    cwd = None
 
     @property
     def has_process(self) -> bool:
@@ -212,6 +214,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
@@ -223,8 +226,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) -> str | None:
+        """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

@@ -218,6 +218,21 @@ def get_stable_start_time(self, recommended: float = 10.0) -> float:
         """
         return recommended
 
+    def resolve_path(self, path: str) -> str | None:
+        """
+        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.