Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 25 additions & 2 deletions ipykernel/debugger.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import os
import re
import stat
import sys
import typing as t
from pathlib import Path
Expand Down Expand Up @@ -304,6 +305,18 @@ async def send_dap_request(self, msg):
return rep


def _is_own_directory(path):
"""Whether path is a directory, and not a symlink, belonging to this user."""
try:
st = os.lstat(path)
except OSError:
return False
if not stat.S_ISDIR(st.st_mode):
return False
getuid = getattr(os, "getuid", None)
return getuid is None or st.st_uid == getuid()


class Debugger:
"""The debugger class."""

Expand Down Expand Up @@ -427,8 +440,16 @@ def start(self):
"""Start the debugger."""
if not self.debugpy_initialized:
tmp_dir = get_tmp_directory()
if not Path(tmp_dir).exists():
try:
Path(tmp_dir).mkdir(mode=0o700, parents=True)
except FileExistsError:
if not _is_own_directory(tmp_dir):
self.log.error(
"Not starting the debugger: %s already exists and is not a"
" directory belonging to this user",
tmp_dir,
)
return False
host, port = self.debugpy_client.get_host_port()
code = "import debugpy;"
code += 'debugpy.listen(("' + host + '",' + port + "))"
Expand Down Expand Up @@ -470,7 +491,9 @@ async def dumpCell(self, message):
code = message["arguments"]["code"]
file_name = get_file_name(code)

with open(file_name, "w", encoding="utf-8") as f: # noqa: ASYNC230
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
fd = os.open(file_name, flags, 0o600)
with open(fd, "w", encoding="utf-8") as f: # noqa: ASYNC230
f.write(code)

return {
Expand Down
36 changes: 36 additions & 0 deletions tests/test_debugger.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import logging
import sys

import pytest
Expand Down Expand Up @@ -368,6 +369,41 @@ def test_convert_to_long_pathname():
_convert_to_long_pathname(__file__)


@pytest.mark.skipif(debugpy is None, reason="requires debugpy")
@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need privileges on windows")
def test_start_rejects_foreign_tmp_directory(tmp_path, monkeypatch):
from ipykernel.debugger import Debugger

# <tempdir>/ipykernel_<pid> is predictable, so on a shared temp directory
# another user can get there first and point it at a path they own.
elsewhere = tmp_path / "elsewhere"
elsewhere.mkdir()
planted = tmp_path / "ipykernel_1"
planted.symlink_to(elsewhere, target_is_directory=True)
monkeypatch.setattr("ipykernel.debugger.get_tmp_directory", lambda: str(planted))

debugger = Debugger(logging.getLogger(__name__), None, lambda event: None, None, None, [])
assert debugger.start() is False


@pytest.mark.skipif(debugpy is None, reason="requires debugpy")
@pytest.mark.skipif(sys.platform == "win32", reason="O_NOFOLLOW is posix only")
async def test_dump_cell_does_not_follow_symlink(tmp_path, monkeypatch):
from ipykernel.debugger import Debugger

outside = tmp_path / "outside.py"
outside.write_text("# untouched\n")
cell_file = tmp_path / "cell.py"
cell_file.symlink_to(outside)
monkeypatch.setenv("IPYKERNEL_CELL_NAME", str(cell_file))

debugger = Debugger(logging.getLogger(__name__), None, lambda event: None, None, None, [])
message = {"seq": 1, "command": "dumpCell", "arguments": {"code": "x = 1\n"}}
with pytest.raises(OSError, match="symbolic link"):
await debugger.dumpCell(message)
assert outside.read_text() == "# untouched\n"


def test_copy_to_globals(kernel_with_debug):
local_var_name = "var"
global_var_name = "var_copy"
Expand Down