don't follow symlinks in the debugger tmp directory (#1575)

Debugger.start only creates its scratch directory when
Path(tmp_dir).exists() is false, and get_tmp_directory returns the
completely predictable <tempdir>/ipykernel_<pid>. Path.exists follows
symlinks, so on a host with a shared temp directory another local user
can reach that name first and point it at a directory they own; the
kernel then skips the mkdir, the 0o700 mode from #1530 never applies,
and dumpCell writes the cell source straight through the link. I went
back to start after that permissions change and noticed the mkdir is
reachable only when the path does not already exist, which is the one
case someone else gets to control. Planting the predictable name as a
symlink locally confirmed it: start carried on, dumpCell dropped the
cell text into the other directory, and a symlink planted under the
Murmur2 cell name turned the same write into an overwrite of a file
outside the temp tree. Creating the directory first and only accepting
an existing one when lstat says it is a real directory owned by this
user closes that, and opening the dumped cell with O_NOFOLLOW keeps the
file write itself from being redirected; the cell file now lands at
0o600 instead of whatever the umask gives, matching the directory around
it. Added two regressions beside the existing debugger tests, both
skipped when debugpy is absent and on Windows where symlinks need
privileges and O_NOFOLLOW does not exist.

Author: JohanMabille

ipykernel/debugger.py

@@ -3,6 +3,7 @@
 import ast
 import os
 import re
+import stat
 import sys
 import typing as t
 from pathlib import Path
@@ -305,6 +306,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."""
 
@@ -428,8 +441,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 + "))"
@@ -471,7 +492,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 {

tests/test_debugger.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 
 import pytest
@@ -431,6 +432,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"