CVE-2024-22415 (Python, published 2024-01-18T21:15Z)
Description: value: jupyter-lsp is a coding assistance tool for JupyterLab (code navigation + hover suggestions + linters + autocompletion + rename) using Language Server Protocol. Installations of jupyter-lsp running in environments without configured file system access control (on the operating system level), and with jupyter-server instances exposed to non-trusted network are vulnerable to unauthorised access and modification of file system beyond the jupyter root directory. This issue has been patched in version 2.2.2 and all users are advised to upgrade. Users unable to upgrade should uninstall jupyter-lsp.
Severity: nan (CVSS v3 base score: 9.8)
Weakness: CWE-22 \u2014 Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal')
The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
Repository: https://github.com/jupyter-lsp/jupyterlab-lsp
Files changed:
python_packages/jupyter_lsp/jupyter_lsp/handlers.py
python_packages/jupyter_lsp/jupyter_lsp/paths.py
python_packages/jupyter_lsp/jupyter_lsp/serverextension.py
python_packages/jupyter_lsp/jupyter_lsp/tests/conftest.py
python_packages/jupyter_lsp/jupyter_lsp/tests/test_auth.py
python_packages/jupyter_lsp/jupyter_lsp/tests/test_paths.py
python_packages/jupyter_lsp/jupyter_lsp/tests/test_virtual_documents_shadow.py
python_packages/jupyter_lsp/jupyter_lsp/virtual_documents_shadow.py
requirements/utest.txt
scripts/bump_versions.py
Commit message: Merge pull request from GHSA-4qhp-652w-c22x

* Add auth decorators, add traversal guard

* Fix mocks resolving most test failures;

`test_listeners` still fails not sure how to fix it

* Address review comments

* add tests for (un)authn'd REST and WebSocket handlers

* Restore old import for 1.x compat, remove a log

* handle advertised jupyter-server 1.x version

* Lint (isort any mypy)

* More tests for paths

---------

Co-authored-by: Nicholas Bollweg <nick.bollweg@gmail.com>

Vulnerable code:
from jupyter_server.base.zmqhandlers import WebSocketHandler, WebSocketMixin
    """Setup tornado websocket to route to language server sessions"""
    def initialize(self, *args, **kwargs):
        super().initialize(*args, **kwargs)

import pathlib
    root_uri = pathlib.Path(root_dir).expanduser().resolve().as_uri()
    nbapp.io_loop.call_later(0, initialize, nbapp, virtual_documents_uri)
        pass
from ..paths import file_uri_to_path, normalized_uri
    # a wargning should be emitted on third failure
from .paths import file_uri_to_path
pluggy<1.0,>=0.12  # Python 3.5 CI Travis, may need update with new pytest releases, see issue 259


Fixed code:
from jupyter_core.utils import ensure_async
from tornado import web
from tornado.websocket import WebSocketHandler

try:
    from jupyter_server.auth.decorator import authorized
except ImportError:

    def authorized(method):  # type: ignore
        """A no-op fallback for `jupyter_server 1.x`"""
        return method


try:
    from jupyter_server.base.websocket import WebSocketMixin
except ImportError:
    from jupyter_server.base.zmqhandlers import WebSocketMixin
AUTH_RESOURCE = "lsp"

    """Setup tornado websocket to route to language server sessions.

    The logic of `get` and `pre_get` methods is derived from jupyter-server ws handlers,
    and should be kept in sync to follow best practice established by upstream; see:
    https://github.com/jupyter-server/jupyter_server/blob/v2.12.5/jupyter_server/services/kernels/websocket.py#L36
    """

    auth_resource = AUTH_RESOURCE
    async def pre_get(self):
        """Handle a pre_get."""
        # authenticate first
        # authenticate the request before opening the websocket
        user = self.current_user
        if user is None:
            self.log.warning("Couldn't authenticate WebSocket connection")
            raise web.HTTPError(403)

        if not hasattr(self, "authorizer"):
            return

        # authorize the user.
        is_authorized = await ensure_async(
            self.authorizer.is_authorized(self, user, "execute", AUTH_RESOURCE)
        )
        if not is_authorized:
            raise web.HTTPError(403)

    async def get(self, *args, **kwargs):
        """Get an event socket."""
        await self.pre_get()
        res = super().get(*args, **kwargs)
        if res is not None:
            await res

    auth_resource = AUTH_RESOURCE
    @web.authenticated
    @authorized
from pathlib import Path
from typing import Union
    root_uri = Path(root_dir).expanduser().resolve().as_uri()


def is_relative(root: Union[str, Path], path: Union[str, Path]) -> bool:
    """Return if path is relative to root"""
    try:
        Path(path).resolve().relative_to(Path(root).resolve())
        return True
    except ValueError:
        return False
from tornado import ioloop

    if hasattr(nbapp, "io_loop"):
        io_loop = nbapp.io_loop
    else:
        # handle jupyter_server 1.x
        io_loop = ioloop.IOLoop.current()

    io_loop.call_later(0, initialize, nbapp, virtual_documents_uri)
from tornado.httpserver import HTTPRequest
    _jupyter_current_user = "foo"  # type:ignore[assignment]
        self.request = HTTPRequest("GET")
        self.application = Application()
"""Integration tests of authorization running under jupyter-server."""
import json
import os
import socket
import subprocess
import time
import uuid
from typing import Generator, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import urlopen

import pytest

from .conftest import KNOWN_SERVERS

LOCALHOST = "127.0.0.1"
REST_ROUTES = ["/lsp/status"]
WS_ROUTES = [f"/lsp/ws/{ls}" for ls in KNOWN_SERVERS]


@pytest.mark.parametrize("route", REST_ROUTES)
def test_auth_rest(route: str, a_server_url_and_token: Tuple[str, str]) -> None:
    """Verify a REST route only provides access to an authenticated user."""
    base_url, token = a_server_url_and_token

    verify_response(base_url, route)

    url = f"{base_url}{route}"

    with urlopen(f"{url}?token={token}") as response:
        raw_body = response.read().decode("utf-8")

    decode_error = None

    try:
        json.loads(raw_body)
    except json.decoder.JSONDecodeError as err:
        decode_error = err
    assert not decode_error, f"the response for {url} was not JSON"


@pytest.mark.parametrize("route", WS_ROUTES)
def test_auth_websocket(route: str, a_server_url_and_token: Tuple[str, str]) -> None:
    """Verify a WebSocket does not provide access to an unauthenticated user."""
    verify_response(a_server_url_and_token[0], route)


@pytest.fixture(scope="module")
def a_server_url_and_token(
    tmp_path_factory: pytest.TempPathFactory,
) -> Generator[Tuple[str, str], None, None]:
    """Start a temporary, isolated jupyter server."""
    token = str(uuid.uuid4())
    port = get_unused_port()

    root_dir = tmp_path_factory.mktemp("root_dir")
    home = tmp_path_factory.mktemp("home")
    server_conf = home / "etc/jupyter/jupyter_config.json"

    server_conf.parent.mkdir(parents=True)
    extensions = {"jupyter_lsp": True, "jupyterlab": False, "nbclassic": False}
    app = {"jpserver_extensions": extensions, "token": token}
    config_data = {"ServerApp": app, "IdentityProvider": {"token": token}}

    server_conf.write_text(json.dumps(config_data), encoding="utf-8")
    args = ["jupyter-server", f"--port={port}", "--no-browser"]
    env = dict(os.environ)
    env.update(
        HOME=str(home),
        USERPROFILE=str(home),
        JUPYTER_CONFIG_DIR=str(server_conf.parent),
    )
    proc = subprocess.Popen(args, cwd=str(root_dir), env=env, stdin=subprocess.PIPE)
    url = f"http://{LOCALHOST}:{port}"
    retries = 20
    while retries:
        time.sleep(1)
        try:
            urlopen(f"{url}/favicon.ico")
            break
        except URLError:
            print(f"[{retries} / 20] ...", flush=True)
            retries -= 1
            continue
    yield url, token
    proc.terminate()
    proc.communicate(b"y\n")
    proc.wait()
    assert proc.returncode is not None, "jupyter-server probably still running"


def get_unused_port():
    """Get an unused port by trying to listen to any random port.

    Probably could introduce race conditions if inside a tight loop.
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind((LOCALHOST, 0))
    sock.listen(1)
    port = sock.getsockname()[1]
    sock.close()
    return port


def verify_response(base_url: str, route: str, expect: int = 403):
    """Verify that a response returns the expected error."""
    error = None
    body = None
    url = f"{base_url}{route}"
    try:
        with urlopen(url) as res:
            body = res.read()
    except HTTPError as err:
        error = err
    assert error, f"no HTTP error for {url}: {body}"
    http_code = error.getcode()
    assert http_code == expect, f"{url} HTTP code was unexpected: {body}"
from ..paths import file_uri_to_path, is_relative, normalized_uri
@pytest.mark.skipif(WIN, reason="can't test POSIX paths on Windows")
@pytest.mark.parametrize(
    "root, path",
    [["~", "~/a"], ["~", "~/a/../b/"], ["/", "/"], ["/a", "/a/b"], ["/a", "/a/b/../c"]],
)
def test_is_relative_ok(root, path):
    assert is_relative(root, path)


@pytest.mark.skipif(WIN, reason="can't test POSIX paths on Windows")
@pytest.mark.parametrize(
    "root, path",
    [
        ["~", "~/.."],
        ["~", "/"],
        ["/a", "/"],
        ["/a/b", "/a"],
        ["/a/b", "/a/b/.."],
        ["/a", "/a/../b"],
        ["/a", "a//"],
    ],
)
def test_is_relative_not_ok(root, path):
    assert not is_relative(root, path)


@pytest.mark.skipif(not WIN, reason="can't test Windows paths on POSIX")
@pytest.mark.parametrize(
    "root, path",
    [
        ["c:\\Users\\user1", "c:\\Users\\"],
        ["c:\\Users\\user1", "d:\\"],
        ["c:\\Users", "c:\\Users\\.."],
    ],
)
def test_is_relative_not_ok_win(root, path):
    assert not is_relative(root, path)


@pytest.mark.asyncio
async def test_shadow_traversal(shadow_path, manager):
    file_beyond_shadow_root_uri = (Path(shadow_path) / ".." / "test.py").as_uri()

    shadow = setup_shadow_filesystem(Path(shadow_path).as_uri())

    def run_shadow(message):
        return shadow("client", message, "python-lsp-server", manager)

    with pytest.raises(
        ShadowFilesystemError, match="is not relative to shadow filesystem root"
    ):
        await run_shadow(did_open(file_beyond_shadow_root_uri, "content"))


    # a warning should be emitted on third failure
from .paths import file_uri_to_path, is_relative
        if not is_relative(shadow_filesystem, path):
            raise ShadowFilesystemError(
                f"Path {path} is not relative to shadow filesystem root"
            )