Path resolver API

jupyter_server/services/api/api.yaml

@@ -352,6 +352,26 @@ paths:
       responses:
         204:
           description: Checkpoint deleted
+  /api/resolvePath:
+    parameters:
+      - name: path
+        required: true
+        in: query
+        description: file path
+        type: string
+      - name: kernel_id
+        required: false
+        in: query
+        description: kernel uuid
+        type: string
+        format: uuid
+    get:
+      summary: Resolve path to a file
+      responses:
+        200:
+          description: Resolved path with scope details
+          schema:
+            $ref: "#/definitions/ResolvedPath"
   /api/sessions/{session}:
     parameters:
       - $ref: "#/parameters/session"
@@ -947,3 +967,24 @@ definitions:
           ISO 8601 timestamp for the last-seen activity on this terminal.  Use
           this to identify which terminals have been inactive since a given time.
           Timestamps will be UTC, indicated 'Z' suffix.
+  ResolvedPath:
+    description: Resolved file path details
+    type: object
+    required:
+      - resolved
+    properties:
+      resolved:
+        type: array
+        description: array of paths to which the query path resolves
+        items:
+          type: object
+          required:
+            - scope
+            - path
+          properties:
+            scope:
+              type: string
+              description: the scope which owns the path
+            path:
+              type: string
+              description: fully specified path for file

jupyter_server/services/api/handlers.py

@@ -105,8 +105,33 @@ def get(self):
         self.write(json.dumps(model))
 
 
+class PathResolverHandler(APIHandler):
+    """Path resolver handler."""
+
+    auth_resource = AUTH_RESOURCE
+    _track_activity = False
+
+    @web.authenticated
+    @authorized
+    async def get(self):
+        """Resolve the path."""
+        path = self.get_query_argument("path")
+        kernel_uuid = self.get_query_argument("kernel", default=None)
+        scopes = {"server": self.contents_manager}
+        if kernel_uuid:
+            scopes["kernel"] = self.kernel_manager.get_kernel(kernel_uuid)
+        resolved = [
+            {"scope": name, "path": await ensure_async(scope.resolve_path(path))}
+            for name, scope in scopes.items()
+            if hasattr(scope, "resolve_path")
+        ]
+        response = {"resolved": [entry for entry in resolved if entry["path"] is not None]}
+        self.finish(json.dumps(response))
+
+
 default_handlers = [
     (r"/api/spec.yaml", APISpecHandler),
     (r"/api/status", APIStatusHandler),
     (r"/api/me", IdentityHandler),
+    (r"/api/resolvePath", PathResolverHandler),
 ]

jupyter_server/services/contents/filemanager.py

@@ -219,6 +219,19 @@ def exists(self, path):
         os_path = self._get_os_path(path=path)
         return exists(os_path)
 
+    def resolve_path(self, path: str):
+        """Resolve path relative to root resource."""
+        # transform OS path to API path
+        relative_path = to_api_path(path, self.root_dir)
+        # check if the API path is within contents directory
+        try:
+            os_path = self._get_os_path(path=relative_path)
+        except web.HTTPError:
+            return None
+        if exists(os_path):
+            return relative_path
+        return None
+
     def _base_model(self, path):
         """Build the common base of a contents model"""
         os_path = self._get_os_path(path)

jupyter_server/services/contents/manager.py

@@ -873,6 +873,10 @@ async def rename_file(self, old_path, new_path):
     # ContentsManager API part 2: methods that have useable default
     # implementations, but can be overridden in subclasses.
 
+    async def resolve_path(self, path: str):
+        """Resolve path relative to root resource."""
+        return
+
     async def delete(self, path):
         """Delete a file/directory and any associated checkpoints."""
         path = path.strip("/")

tests/services/kernels/test_api.py

@@ -183,6 +183,153 @@ async def test_main_kernel_handler(jp_fetch, jp_base_url, jp_serverapp, pending_
     await pending_kernel_is_ready(kernel3["id"])
 
 
+@pytest.mark.timeout(TEST_TIMEOUT)
+async def test_resolve_path_kernel(jp_fetch, jp_serverapp, jp_root_dir):
+    query_path = "hello.py"
+    contents = "(irrelevant)"
+
+    # Put the file in a kernel subdirectory
+    kernel_root = jp_root_dir / "more" / "path" / "segments"
+    kernel_root.mkdir(parents=True)
+    kernel_path = kernel_root / query_path
+    kernel_path.write_text(contents)
+
+    # Create a kernel in temp path
+    r = await jp_fetch(
+        "api",
+        "kernels",
+        method="POST",
+        body=json.dumps(
+            {"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
+        ),
+    )
+    kernel_id = json.loads(r.body.decode())["id"]
+
+    # Resolve the path
+    r = await jp_fetch(
+        "api", "resolvePath", params={"kernel": kernel_id, "path": query_path}, method="GET"
+    )
+
+    assert r.code == 200
+    resolution = json.loads(r.body.decode())
+    assert len(resolution["resolved"]) == 1
+    path = resolution["resolved"][0]
+    assert path["scope"] == "kernel"
+
+    # Should reveal the kernel path
+    assert path["path"] == str(kernel_path)
+
+
+@pytest.mark.timeout(TEST_TIMEOUT)
+async def test_resolve_path_server_traversal(jp_fetch, jp_serverapp, jp_root_dir):
+    query_path = "hello.py"
+    contents = "(irrelevant)"
+
+    # Put the file above the root dir
+    server_path = jp_root_dir / ".." / query_path
+    server_path.write_text(contents)
+
+    # Resolve the path
+    r = await jp_fetch("api", "resolvePath", params={"path": query_path}, method="GET")
+
+    assert r.code == 200
+    resolution = json.loads(r.body.decode())
+
+    # Should NOT disclose file existence on path traversal
+    assert len(resolution["resolved"]) == 0
+
+
+@pytest.mark.timeout(TEST_TIMEOUT)
+async def test_resolve_path_server(jp_fetch, jp_serverapp, jp_root_dir):
+    query_path = "hello.py"
+    contents = "(irrelevant)"
+
+    # Put the file in the root dir
+    server_path = jp_root_dir / query_path
+    server_path.write_text(contents)
+
+    # Resolve the path
+    r = await jp_fetch("api", "resolvePath", params={"path": query_path}, method="GET")
+
+    assert r.code == 200
+    resolution = json.loads(r.body.decode())
+    assert len(resolution["resolved"]) == 1
+    path = resolution["resolved"][0]
+    assert path["scope"] == "server"
+
+    # Should NOT reveal server path, just acknowledge the name
+    assert path["path"] == server_path.name
+
+
+@pytest.mark.timeout(TEST_TIMEOUT)
+async def test_resolve_path_server_and_kernel(jp_fetch, jp_serverapp, jp_root_dir):
+    query_path = "hello.py"
+    contents = "(irrelevant)"
+
+    # Put a file for server in the root and for kernel space in a subdirectory
+    kernel_root = jp_root_dir / "more" / "path" / "segments"
+    kernel_root.mkdir(parents=True)
+    kernel_path = kernel_root / query_path
+    kernel_path.write_text(contents)
+    server_path = jp_root_dir / query_path
+    server_path.write_text(contents)
+
+    # Create a kernel in temp path
+    r = await jp_fetch(
+        "api",
+        "kernels",
+        method="POST",
+        body=json.dumps(
+            {"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
+        ),
+    )
+    kernel_id = json.loads(r.body.decode())["id"]
+
+    # Resolve the path
+    r = await jp_fetch(
+        "api", "resolvePath", params={"kernel": kernel_id, "path": query_path}, method="GET"
+    )
+    assert r.code == 200
+    resolution = json.loads(r.body.decode())
+
+    # Should present candidates for both server and kernel
+    assert len(resolution["resolved"]) == 2
+    assert {k["scope"] for k in resolution["resolved"]} == {"server", "kernel"}
+
+
+@pytest.mark.timeout(TEST_TIMEOUT)
+async def test_resolve_path_missing(jp_fetch, jp_serverapp, jp_root_dir):
+    query_path = "hello.py"
+    contents = "(irrelevant)"
+
+    # Put a file for server in the root and for kernel space in a subdirectory
+    kernel_root = jp_root_dir / "more" / "path" / "segments"
+    kernel_root.mkdir(parents=True)
+    kernel_path = kernel_root / query_path
+    kernel_path.write_text(contents)
+    server_path = jp_root_dir / query_path
+    server_path.write_text(contents)
+
+    # Create a kernel in temp path
+    r = await jp_fetch(
+        "api",
+        "kernels",
+        method="POST",
+        body=json.dumps(
+            {"name": NATIVE_KERNEL_NAME, "path": str(kernel_root.relative_to(jp_root_dir))}
+        ),
+    )
+    kernel_id = json.loads(r.body.decode())["id"]
+
+    # Resolve the path
+    r = await jp_fetch(
+        "api", "resolvePath", params={"kernel": kernel_id, "path": "not_hello"}, method="GET"
+    )
+    assert r.code == 200
+    resolution = json.loads(r.body.decode())
+    assert len(resolution["resolved"]) == 0
+
+
 @pytest.mark.timeout(TEST_TIMEOUT)
 async def test_kernel_handler(jp_fetch, jp_serverapp, pending_kernel_is_ready):
     # Create a kernel