diff --git a/docs/changelog/213.security.rst b/docs/changelog/213.security.rst new file mode 100644 index 0000000..28a3d2f --- /dev/null +++ b/docs/changelog/213.security.rst @@ -0,0 +1 @@ +Skip untrusted disk cache directories and entries on POSIX, and replace cache files without following links. diff --git a/docs/explanation.rst b/docs/explanation.rst index 2b1ca5e..cf4869c 100644 --- a/docs/explanation.rst +++ b/docs/explanation.rst @@ -263,10 +263,18 @@ with `filelock `_-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 ---------------------------- diff --git a/src/python_discovery/_cache.py b/src/python_discovery/_cache.py index 86797f3..b0eeb8c 100644 --- a/src/python_discovery/_cache.py +++ b/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): diff --git a/tests/test_cache.py b/tests/test_cache.py index e280877..933d44f 100644 --- a/tests/test_cache.py +++ b/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