🐛 fix(cache): reject untrusted cache paths

docs/changelog/213.security.rst

@@ -0,0 +1 @@
+Skip untrusted disk cache directories and entries on POSIX, and replace cache files without following links.

docs/explanation.rst

@@ -263,10 +263,18 @@ with `filelock <https://py-filelock.readthedocs.io/>`_-based locking for safe co
 can also pass ``cache=None`` to disable caching, or implement your own backend (see
 :doc:`/how-to/standalone-usage`).
 
-On POSIX, lock files live in a restricted directory accessible to the cache owner. When the cache entry directory is
-writable by its group and the process uses a group-permissive umask, the lock directory and files allow group writes
-too. Older releases still use the previous lock files, so processes running different releases do not coordinate cache
-writes. Use separate cache roots while both run.
+On POSIX, ``DiskCache`` uses a cache root only when it and both ``py_info`` directories are real directories owned by
+the current user or root and cannot be written by other users. Its ancestor directories must also prevent other users
+from replacing the cache root; a sticky shared ancestor such as ``/tmp`` is allowed. Discovery continues without disk
+caching and logs a warning if the directory check fails. Cache entries must be regular files owned by the current user
+or root and not writable by others. Discovery tries to remove a rejected entry and queries the interpreter again. Use a
+private directory for the cache root, including when callers share a machine.
+
+A relative cache root refers to the current directory when ``DiskCache`` is created.
+
+On POSIX, lock files live in a restricted directory under the checked cache directory. Older releases use the previous
+lock files, so processes running different releases do not coordinate cache writes. Use separate cache roots while both
+run.
 
 Subprocess timeout behavior
 ----------------------------

src/python_discovery/_cache.py

@@ -6,13 +6,14 @@
 import logging
 import os
 import stat
+import tempfile
 from contextlib import contextmanager, suppress
 from hashlib import sha256
+from pathlib import Path
 from typing import TYPE_CHECKING, Final, Protocol, runtime_checkable
 
 if TYPE_CHECKING:
     from collections.abc import Generator
-    from pathlib import Path
 
 _LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
 
@@ -81,23 +82,53 @@ def exists(self) -> bool:
     def read(self) -> dict | None:
         data, bad_format = None, False
         try:
-            data = json.loads(self._file.read_text(encoding="utf-8"))
-        except ValueError:
+            data, bad_format = self._read_entry()
+        except (ValueError, RecursionError):
             bad_format = True
         except OSError:
             _LOGGER.debug("failed to read %s", self._file, exc_info=True)
+            bad_format = self._file.is_symlink()
         else:
-            _LOGGER.debug("got python info from %s", self._file)
-            return data
+            if not bad_format:
+                _LOGGER.debug("got python info from %s", self._file)
+                return data
         if bad_format:
             with suppress(OSError):
                 self.remove()
         return None
 
+    def _read_entry(self) -> tuple[dict | None, bool]:
+        with open(self._file, encoding="utf-8", opener=self._open_entry) as file:
+            status = os.fstat(file.fileno())
+            if not stat.S_ISREG(status.st_mode) or (
+                hasattr(os, "geteuid") and (status.st_uid not in {os.geteuid(), 0} or status.st_mode & 0o022)
+            ):
+                _LOGGER.warning("ignoring untrusted cache entry at %s", self._file)
+                return None, True
+            return json.load(file), False
+
+    @staticmethod
+    def _open_entry(path: str, flags: int) -> int:
+        return os.open(path, flags | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
+
     def write(self, content: dict) -> None:
         self._folder.mkdir(parents=True, exist_ok=True)
-        self._file.write_text(json.dumps(content, sort_keys=True, indent=2), encoding="utf-8")
-        _LOGGER.debug("wrote python info at %s", self._file)
+        temporary: Path | None = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", encoding="utf-8", dir=self._folder, prefix=f".{self._key}.", delete=False
+            ) as file:
+                temporary = Path(file.name)
+                json.dump(content, file, sort_keys=True, indent=2)
+            temporary.replace(self._file)
+        except OSError:
+            _LOGGER.debug("failed to write %s", self._file, exc_info=True)
+        else:
+            _LOGGER.debug("wrote python info at %s", self._file)
+        finally:
+            if temporary is not None:
+                with suppress(OSError):
+                    temporary.unlink(missing_ok=True)
 
     def remove(self) -> None:
         with suppress(OSError):
@@ -128,25 +159,52 @@ class DiskCache:
     """
 
     def __init__(self, root: Path) -> None:
-        self._root = root
+        self._root = Path(os.path.abspath(root))  # ruff:ignore[os-path-abspath]  # resolve() follows untrusted links
 
     @property
     def _py_info_dir(self) -> Path:
         return self._root / "py_info" / "4"
 
-    def py_info(self, path: Path) -> DiskContentStore:
+    def py_info(self, path: Path) -> ContentStore:
         """
         Return the content store for the interpreter at *path*.
 
         :param path: absolute path to a Python executable.
         """
+        if not self._trusted_folder():
+            return NoOpContentStore()
         key = sha256(str(path).encode("utf-8")).hexdigest()
         return DiskContentStore(self._py_info_dir, key)
 
+    def _trusted_folder(self) -> bool:
+        if not hasattr(os, "geteuid"):
+            return True
+        user_id = os.geteuid()
+        folder = self._py_info_dir
+        private = {self._root, folder.parent, folder}
+        for candidate in (*reversed(folder.parents), folder):
+            try:
+                try:
+                    status = candidate.lstat()
+                except FileNotFoundError:
+                    candidate.mkdir(exist_ok=True)
+                    status = candidate.lstat()
+            except OSError:
+                _LOGGER.warning("not caching in %s", candidate, exc_info=True)
+                return False
+            if (
+                not stat.S_ISDIR(status.st_mode)
+                or status.st_uid not in {user_id, 0}
+                or (status.st_mode & 0o022 and (candidate in private or not status.st_mode & stat.S_ISVTX))
+            ):
+                _LOGGER.warning("not caching in untrusted directory %s", candidate)
+                return False
+        return True
+
     def py_info_clear(self) -> None:
         """Remove all cached interpreter information."""
         folder = self._py_info_dir
-        if folder.exists():
+        if folder.exists() and self._trusted_folder():
             for entry in folder.iterdir():
                 if entry.suffix == ".json":
                     with suppress(OSError):

tests/test_cache.py

@@ -145,6 +145,162 @@ def test_disk_cache_py_info_clear_skips_non_json(tmp_path: Path) -> None:
     assert "test.json" not in remaining
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+@pytest.mark.parametrize("unsafe", [pytest.param("", id="root"), pytest.param("py_info/4", id="entry-dir")])
+def test_disk_cache_skips_writable_directory(tmp_path: Path, unsafe: str) -> None:
+    folder = tmp_path / unsafe
+    folder.mkdir(parents=True, exist_ok=True)
+    folder.chmod(0o777)
+    assert isinstance(DiskCache(tmp_path).py_info(Path("/some/path")), NoOpContentStore)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_skips_linked_directory(tmp_path: Path) -> None:
+    target = tmp_path / "target"
+    target.mkdir()
+    (tmp_path / "py_info").symlink_to(target, target_is_directory=True)
+    assert isinstance(DiskCache(tmp_path).py_info(Path("/some/path")), NoOpContentStore)
+    assert not (target / "4").exists()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_skips_writable_ancestor(tmp_path: Path) -> None:
+    ancestor = tmp_path / "shared"
+    ancestor.mkdir()
+    ancestor.chmod(0o777)
+    assert isinstance(DiskCache(ancestor / "cache").py_info(Path("/some/path")), NoOpContentStore)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_skips_file_as_root(tmp_path: Path) -> None:
+    root = tmp_path / "file"
+    root.touch()
+    assert isinstance(DiskCache(root).py_info(Path("/some/path")), NoOpContentStore)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_skips_uncreatable_directory(tmp_path: Path, mocker: MockerFixture) -> None:
+    mocker.patch.object(Path, "mkdir", side_effect=PermissionError)
+    assert isinstance(DiskCache(tmp_path / "cache").py_info(Path("/some/path")), NoOpContentStore)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_skips_foreign_directory(tmp_path: Path, mocker: MockerFixture) -> None:
+    status = list(tmp_path.stat())
+    status[4] = os.geteuid() + 1
+    mocker.patch.object(Path, "lstat", return_value=os.stat_result(status))
+    assert isinstance(DiskCache(tmp_path).py_info(Path("/some/path")), NoOpContentStore)
+
+
+def test_disk_cache_works_without_posix_ownership(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delattr(os, "geteuid", raising=False)
+    assert isinstance(DiskCache(tmp_path).py_info(Path("/some/path")), DiskContentStore)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_keeps_validated_relative_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    safe = tmp_path / "safe"
+    unsafe = tmp_path / "unsafe"
+    safe.mkdir()
+    unsafe.mkdir()
+    unsafe.chmod(0o777)
+    monkeypatch.chdir(safe)
+    store = DiskCache(Path("cache")).py_info(Path("/some/path"))
+    monkeypatch.chdir(unsafe)
+    store.write({"key": "value"})
+    assert (len(list(safe.rglob("*.json"))), list(unsafe.rglob("*.json"))) == (1, [])
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+def test_disk_cache_clear_skips_writable_directory(tmp_path: Path) -> None:
+    folder = tmp_path / "py_info" / "4"
+    folder.mkdir(parents=True)
+    entry = folder / "entry.json"
+    entry.write_text("{}", encoding="utf-8")
+    tmp_path.chmod(0o777)
+    DiskCache(tmp_path).py_info_clear()
+    assert entry.exists()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symbolic links")
+def test_disk_cache_clear_does_not_follow_linked_directory(tmp_path: Path) -> None:
+    target = tmp_path / "target" / "4"
+    target.mkdir(parents=True)
+    entry = target / "credentials.json"
+    entry.write_text("{}", encoding="utf-8")
+    (tmp_path / "py_info").symlink_to(target.parent, target_is_directory=True)
+    DiskCache(tmp_path).py_info_clear()
+    assert entry.exists()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX ownership and modes")
+@pytest.mark.parametrize(
+    ("owner_offset", "mode"),
+    [pytest.param(1, 0o600, id="foreign-owner"), pytest.param(0, 0o660, id="group-writable")],
+)
+def test_disk_content_store_rejects_untrusted_entry(
+    tmp_path: Path, mocker: MockerFixture, owner_offset: int, mode: int
+) -> None:
+    store = DiskContentStore(tmp_path, "entry")
+    store.write({"key": "value"})
+    status = list((tmp_path / "entry.json").stat())
+    status[0], status[4] = stat.S_IFREG | mode, os.geteuid() + owner_offset
+    mocker.patch("python_discovery._cache.os.fstat", return_value=os.stat_result(status))
+    assert store.read() is None
+    assert not store.exists()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symbolic links")
+def test_disk_content_store_does_not_follow_entry_link(tmp_path: Path) -> None:
+    target = tmp_path / "target.json"
+    target.write_text('{"key": "target"}', encoding="utf-8")
+    entry = tmp_path / "entry.json"
+    entry.symlink_to(target)
+    store = DiskContentStore(tmp_path, "entry")
+    assert store.read() is None
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX FIFOs")
+def test_disk_content_store_skips_fifo(tmp_path: Path) -> None:
+    os.mkfifo(tmp_path / "entry.json")
+    store = DiskContentStore(tmp_path, "entry")
+    assert store.read() is None
+    assert not store.exists()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symbolic links")
+def test_disk_content_store_write_replaces_entry_link(tmp_path: Path) -> None:
+    target = tmp_path / "target.json"
+    target.write_text('{"key": "target"}', encoding="utf-8")
+    (tmp_path / "entry.json").symlink_to(target)
+    store = DiskContentStore(tmp_path, "entry")
+    store.write({"key": "value"})
+    assert (target.read_text(encoding="utf-8"), store.read()) == ('{"key": "target"}', {"key": "value"})
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX file modes")
+def test_disk_content_store_write_is_private_with_lax_umask(tmp_path: Path) -> None:
+    previous = os.umask(0)
+    try:
+        DiskContentStore(tmp_path, "entry").write({"key": "value"})
+    finally:
+        os.umask(previous)
+    assert stat.S_IMODE((tmp_path / "entry.json").stat().st_mode) == 0o600
+
+
+@pytest.mark.parametrize(
+    "target",
+    [
+        pytest.param("python_discovery._cache.tempfile.NamedTemporaryFile", id="create"),
+        pytest.param("pathlib.Path.replace", id="replace"),
+    ],
+)
+def test_disk_content_store_write_skips_io_failure(tmp_path: Path, mocker: MockerFixture, target: str) -> None:
+    mocker.patch(target, side_effect=PermissionError)
+    DiskContentStore(tmp_path, "entry").write({"key": "value"})
+    assert list(tmp_path.iterdir()) == []
+
+
 def test_noop_content_store_exists() -> None:
     assert NoOpContentStore().exists() is False