diff --git a/jupyter_client/manager.py b/jupyter_client/manager.py index a668bc06..59fa817f 100644 --- a/jupyter_client/manager.py +++ b/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. diff --git a/jupyter_client/provisioning/local_provisioner.py b/jupyter_client/provisioning/local_provisioner.py index 9f1086f1..20e4802f 100644 --- a/jupyter_client/provisioning/local_provisioner.py +++ b/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.""" diff --git a/jupyter_client/provisioning/provisioner_base.py b/jupyter_client/provisioning/provisioner_base.py index 8056a1ca..3e93f81d 100644 --- a/jupyter_client/provisioning/provisioner_base.py +++ b/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.