🐛 fix(cache): move locks to private directory

docs/changelog/214.security.rst

@@ -0,0 +1 @@
+Place POSIX cache locks in a restricted directory and preserve group sharing under a group-permissive umask.

docs/explanation.rst

@@ -258,6 +258,11 @@ 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.
+
 Subprocess timeout behavior
 ----------------------------
 

src/python_discovery/_cache.py

@@ -4,6 +4,8 @@
 
 import json
 import logging
+import os
+import stat
 from contextlib import contextmanager, suppress
 from hashlib import sha256
 from typing import TYPE_CHECKING, Final, Protocol, runtime_checkable
@@ -106,9 +108,15 @@ def remove(self) -> None:
     def locked(self) -> Generator[None]:
         from filelock import FileLock  # ruff:ignore[import-outside-top-level]
 
-        lock_path = self._folder / f"{self._key}.lock"
-        lock_path.parent.mkdir(parents=True, exist_ok=True)
-        with FileLock(str(lock_path)):
+        self._folder.mkdir(parents=True, exist_ok=True)
+        if os.name == "posix":
+            group_shared = bool(self._folder.stat().st_mode & stat.S_IWGRP)
+            lock_folder = self._folder / "locks-v2"
+            lock_folder.mkdir(mode=0o770 if group_shared else 0o700, exist_ok=True)
+            lock = FileLock(str(lock_folder / f"{self._key}.lock"), mode=0o660 if group_shared else 0o600)
+        else:
+            lock = FileLock(str(self._folder / f"{self._key}.lock"))
+        with lock:
             yield
 
 

tests/test_cache.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+import os
+import stat
+import sys
 from pathlib import Path
+from threading import Event, Thread
+from typing import TYPE_CHECKING
+
+import pytest
+from filelock import FileLock
 
 from python_discovery._cache import DiskCache, DiskContentStore, NoOpCache, NoOpContentStore
 
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
 
 def test_disk_content_store_read_valid_json(tmp_path: Path) -> None:
     store = DiskContentStore(tmp_path, "test")
@@ -45,6 +56,56 @@ def test_disk_content_store_locked(tmp_path: Path) -> None:
     assert store.read() == {"key": "value"}
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX lock permissions")
+@pytest.mark.parametrize(
+    ("folder_mode", "umask", "lock_folder_mode", "lock_mode"),
+    [
+        pytest.param(0o755, 0o022, 0o700, 0o600, id="private"),
+        pytest.param(0o770, 0o002, 0o770, 0o660, id="group-shared"),
+    ],
+)
+def test_disk_content_store_lock_permissions(
+    tmp_path: Path, folder_mode: int, umask: int, lock_folder_mode: int, lock_mode: int
+) -> None:
+    tmp_path.chmod(folder_mode)
+    previous = os.umask(umask)
+    try:
+        with DiskContentStore(tmp_path, "test").locked():
+            pass
+    finally:
+        os.umask(previous)
+    assert (
+        stat.S_IMODE((tmp_path / "locks-v2").stat().st_mode),
+        stat.S_IMODE((tmp_path / "locks-v2" / "test.lock").stat().st_mode),
+    ) == (lock_folder_mode, lock_mode)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX lock migration")
+def test_disk_content_store_ignores_legacy_lock(tmp_path: Path) -> None:
+    entered = Event()
+    store = DiskContentStore(tmp_path, "test")
+
+    def acquire() -> None:
+        with store.locked():
+            entered.set()
+
+    thread = Thread(target=acquire)
+    with FileLock(str(tmp_path / "test.lock")):
+        thread.start()
+        acquired = entered.wait(3)
+    thread.join(timeout=3)
+    assert acquired
+
+
+def test_disk_content_store_keeps_lock_path_without_posix(tmp_path: Path, mocker: MockerFixture) -> None:
+    platform_os = mocker.patch("python_discovery._cache.os", autospec=True)
+    platform_os.name = "nt"
+    with DiskContentStore(tmp_path, "test").locked():
+        pass
+    assert (tmp_path / "test.lock").exists()
+    assert not (tmp_path / "locks-v2").exists()
+
+
 def test_disk_content_store_exists(tmp_path: Path) -> None:
     store = DiskContentStore(tmp_path, "test")
     assert store.exists() is False