diff --git a/jupyter_client/manager.py b/jupyter_client/manager.py index 088acd6c..623f9ce5 100644 --- a/jupyter_client/manager.py +++ b/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. diff --git a/jupyter_client/provisioning/local_provisioner.py b/jupyter_client/provisioning/local_provisioner.py index 42d8d32d..bf3369c7 100644 --- a/jupyter_client/provisioning/local_provisioner.py +++ b/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.""" diff --git a/jupyter_client/provisioning/provisioner_base.py b/jupyter_client/provisioning/provisioner_base.py index eff89432..d90c68a2 100644 --- a/jupyter_client/provisioning/provisioner_base.py +++ b/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.