CVE-2024-22415: Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal') (CWE-22)
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.
CWE description: 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.
CVSS v3 base score: 9.8
Published: 2024-01-18T21:15Z
Repository: https://github.com/jupyter-lsp/jupyterlab-lsp
Language: Python
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 date: 2024-01-17 21:58:35 +0000
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>
Diff: diff --git a/python_packages/jupyter_lsp/jupyter_lsp/handlers.py b/python_packages/jupyter_lsp/jupyter_lsp/handlers.py
index 0ff6ff0b..75f2eb4c 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/handlers.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/handlers.py
@@ -1,17 +1,35 @@
 """ tornado handler for managing and communicating with language servers
 """
 from typing import Optional, Text
 
+from jupyter_core.utils import ensure_async
 from jupyter_server.base.handlers import APIHandler
-from jupyter_server.base.zmqhandlers import WebSocketHandler, WebSocketMixin
 from jupyter_server.utils import url_path_join as ujoin
+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
 
 from .manager import LanguageServerManager
 from .schema import SERVERS_RESPONSE
 from .specs.utils import censored_spec
 
+AUTH_RESOURCE = "lsp"
+
 
 class BaseHandler(APIHandler):
     manager = None  # type: LanguageServerManager
 
     def initialize(self, manager: LanguageServerManager):
@@ -19,14 +37,47 @@ class BaseHandler(APIHandler):
 
 
 class LanguageServerWebSocketHandler(  # type: ignore
     WebSocketMixin, WebSocketHandler, BaseHandler
 ):
-    """Setup tornado websocket to route to language server sessions"""
+    """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
 
     language_server: Optional[Text] = None
 
+    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
+
     async def open(self, language_server):
         await self.manager.ready()
         self.language_server = language_server
         self.manager.subscribe(self)
         self.log.debug("[{}] Opened a handler".format(self.language_server))
@@ -45,15 +96,15 @@ class LanguageServersHandler(BaseHandler):
     """Reports the status of all current servers
 
     Response should conform to schema in schema/servers.schema.json
     """
 
+    auth_resource = AUTH_RESOURCE
     validator = SERVERS_RESPONSE
 
-    def initialize(self, *args, **kwargs):
-        super().initialize(*args, **kwargs)
-
+    @web.authenticated
+    @authorized
     async def get(self):
         """finish with the JSON representations of the sessions"""
         await self.manager.ready()
 
         response = {
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/paths.py b/python_packages/jupyter_lsp/jupyter_lsp/paths.py
index 5c9cb43e..625226dc 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/paths.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/paths.py
@@ -1,8 +1,9 @@
 import os
-import pathlib
 import re
+from pathlib import Path
+from typing import Union
 from urllib.parse import unquote, urlparse
 
 RE_PATH_ANCHOR = r"^file://([^/]+|/[A-Z]:)"
 
 
@@ -10,11 +11,11 @@ def normalized_uri(root_dir):
     """Attempt to make an LSP rootUri from a ContentsManager root_dir
 
     Special care must be taken around windows paths: the canonical form of
     windows drives and UNC paths is lower case
     """
-    root_uri = pathlib.Path(root_dir).expanduser().resolve().as_uri()
+    root_uri = Path(root_dir).expanduser().resolve().as_uri()
     root_uri = re.sub(
         RE_PATH_ANCHOR, lambda m: "file://{}".format(m.group(1).lower()), root_uri
     )
     return root_uri
 
@@ -31,5 +32,14 @@ def file_uri_to_path(file_uri):
     if windows_path and file_uri_path_unquoted.startswith("/"):
         result = file_uri_path_unquoted[1:]  # pragma: no cover
     else:
         result = file_uri_path_unquoted  # pragma: no cover
     return result
+
+
+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
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/serverextension.py b/python_packages/jupyter_lsp/jupyter_lsp/serverextension.py
index 2640e429..194e9efe 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/serverextension.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/serverextension.py
@@ -2,10 +2,11 @@
 """
 import json
 from pathlib import Path
 
 import traitlets
+from tornado import ioloop
 
 from .handlers import add_handlers
 from .manager import LanguageServerManager
 from .paths import normalized_uri
 
@@ -71,6 +72,13 @@ def load_jupyter_server_extension(nbapp):
             contents,
         )
     page_config.update(rootUri=root_uri, virtualDocumentsUri=virtual_documents_uri)
 
     add_handlers(nbapp)
-    nbapp.io_loop.call_later(0, initialize, nbapp, virtual_documents_uri)
+
+    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)
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/tests/conftest.py b/python_packages/jupyter_lsp/jupyter_lsp/tests/conftest.py
index fd8e0b93..71109a94 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/tests/conftest.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/tests/conftest.py
@@ -3,10 +3,11 @@ import pathlib
 import shutil
 from typing import Text
 
 from jupyter_server.serverapp import ServerApp
 from pytest import fixture
+from tornado.httpserver import HTTPRequest
 from tornado.httputil import HTTPServerRequest
 from tornado.queues import Queue
 from tornado.web import Application
 
 # local imports
@@ -139,13 +140,15 @@ class MockWebsocketHandler(LanguageServerWebSocketHandler):
         self._ping_sent = True
 
 
 class MockHandler(LanguageServersHandler):
     _payload = None
+    _jupyter_current_user = "foo"  # type:ignore[assignment]
 
     def __init__(self):
-        pass
+        self.request = HTTPRequest("GET")
+        self.application = Application()
 
     def finish(self, payload):
         self._payload = payload
 
 
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/tests/test_auth.py b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_auth.py
new file mode 100644
index 00000000..0c0b2e88
--- /dev/null
+++ b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_auth.py
@@ -0,0 +1,117 @@
+"""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}"
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/tests/test_paths.py b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_paths.py
index 3360da4e..6c91d16d 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/tests/test_paths.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_paths.py
@@ -2,11 +2,11 @@ import pathlib
 import platform
 import sys
 
 import pytest
 
-from ..paths import file_uri_to_path, normalized_uri
+from ..paths import file_uri_to_path, is_relative, normalized_uri
 
 WIN = platform.system() == "Windows"
 HOME = pathlib.Path("~").expanduser()
 PY35 = sys.version_info[:2] == (3, 5)
 
@@ -15,10 +15,49 @@ PY35 = sys.version_info[:2] == (3, 5)
 @pytest.mark.parametrize("root_dir, expected_root_uri", [["~", HOME.as_uri()]])
 def test_normalize_posix_path_home(root_dir, expected_root_uri):  # pragma: no cover
     assert normalized_uri(root_dir) == expected_root_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.skipif(PY35, reason="can't test non-existent paths on py35")
 @pytest.mark.skipif(WIN, reason="can't test POSIX paths on Windows")
 @pytest.mark.parametrize(
     "root_dir, expected_root_uri",
     [
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/tests/test_virtual_documents_shadow.py b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_virtual_documents_shadow.py
index 5cce1ac6..0d4c4b53 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/tests/test_virtual_documents_shadow.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/tests/test_virtual_documents_shadow.py
@@ -208,10 +208,25 @@ async def test_shadow_failures(shadow_path, manager):
                 "params": {"textDocument": {"uri": ok_file_uri}},
             }
         )
 
 
+@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"))
+
+
 @pytest.fixture
 def forbidden_shadow_path(tmpdir):
     path = Path(tmpdir) / "no_permission_dir"
     path.mkdir()
     path.chmod(0o000)
@@ -236,11 +251,11 @@ async def test_io_failure(forbidden_shadow_path, manager, caplog):
         assert await send_change() is None
         assert await send_change() is None
     # no message should be emitted during the first two attempts
     assert caplog.text == ""
 
-    # a wargning should be emitted on third failure
+    # a warning should be emitted on third failure
     with caplog.at_level(logging.WARNING):
         assert await send_change() is None
     assert "initialization of shadow filesystem failed three times" in caplog.text
     assert "PermissionError" in caplog.text
     caplog.clear()
diff --git a/python_packages/jupyter_lsp/jupyter_lsp/virtual_documents_shadow.py b/python_packages/jupyter_lsp/jupyter_lsp/virtual_documents_shadow.py
index abdf22ae..b78a3130 100644
--- a/python_packages/jupyter_lsp/jupyter_lsp/virtual_documents_shadow.py
+++ b/python_packages/jupyter_lsp/jupyter_lsp/virtual_documents_shadow.py
@@ -6,11 +6,11 @@ from typing import List
 
 from tornado.concurrent import run_on_executor
 from tornado.gen import convert_yielded
 
 from .manager import lsp_message_listener
-from .paths import file_uri_to_path
+from .paths import file_uri_to_path, is_relative
 from .types import LanguageServerManagerAPI
 
 # TODO: make configurable
 MAX_WORKERS = 4
 
@@ -169,10 +169,15 @@ def setup_shadow_filesystem(virtual_documents_uri: str):
                     )
                 return
             initialized = True
 
         path = file_uri_to_path(uri)
+        if not is_relative(shadow_filesystem, path):
+            raise ShadowFilesystemError(
+                f"Path {path} is not relative to shadow filesystem root"
+            )
+
         editable_file = EditableFile(path)
 
         await editable_file.read()
 
         text = extract_or_none(document, ["text"])
diff --git a/requirements/utest.txt b/requirements/utest.txt
index 726fccae..9a8dd481 100644
--- a/requirements/utest.txt
+++ b/requirements/utest.txt
@@ -4,6 +4,5 @@ flake8 >=3.5
 pytest-asyncio
 pytest-cov
 pytest-flake8
 pytest-runner
 python-lsp-server
-pluggy<1.0,>=0.12  # Python 3.5 CI Travis, may need update with new pytest releases, see issue 259
diff --git a/scripts/bump_versions.py b/scripts/bump_versions.py
index a70c73f3..28eef97c 100755
--- a/scripts/bump_versions.py
+++ b/scripts/bump_versions.py
@@ -45,11 +45,10 @@ class PackageVersionInfo:
         version = input("Change it to [default=skip]: ").strip()
         if version:
             self.change_version(new_version=version, dry=dry)
 
     def change_version(self, new_version: str, dry: bool):
-
         changelog = CHANGELOG.read_text(encoding="utf-8")
         if new_version not in changelog:
             raise Exception(
                 (
                     f"{new_version} is absent in CHANGELOG.md file."

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"
            )