tig-cli: keep container claims in a private, user-owned directory (SonarCloud S5443)

tig-cli/src/tig_cli/spec.py

@@ -11,6 +11,7 @@
 import json
 import os
 import signal
+import stat
 import sys
 from pathlib import Path
 from .config import (
@@ -658,20 +659,53 @@ def __init__(self) -> None:
 
     @staticmethod
     def directory() -> Path:
-        base = (
-            os.environ.get("XDG_RUNTIME_DIR")
-            or os.environ.get("TMPDIR")
-            or "/tmp"
-        )
+        base = os.environ.get("XDG_RUNTIME_DIR")
+        if not base:
+            # Imported only here: the warm path almost always has a runtime
+            # directory, and tempfile is comparatively slow to import.
+            import tempfile
+
+            base = tempfile.gettempdir()
         return Path(base) / f"{CLAIM_DIR_NAME}-{os.getuid()}"
 
+    @classmethod
+    def _private_directory(cls, create: bool) -> Path | None:
+        """The claim directory, if it is safe to trust.
+
+        The fallback base is shared and world-writable, so another user could
+        have made the directory first, or a symlink in its place, to plant or
+        delete claims. Only a real directory owned by this user and closed to
+        everyone else is used.
+        """
+        directory = cls.directory()
+        if create:
+            try:
+                directory.mkdir(mode=0o700, parents=True)
+            except FileExistsError:
+                pass
+            except OSError:
+                return None
+        try:
+            info = os.lstat(directory)
+        except OSError:
+            return None
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
+            return None
+        if stat.S_IMODE(info.st_mode) & 0o077:
+            try:
+                os.chmod(directory, 0o700)
+            except OSError:
+                return None
+        return directory
+
     def acquire(self, container_name: str) -> None:
         """Record that this process is about to use ``container_name``."""
         self.release()
-        directory = self.directory()
+        directory = self._private_directory(create=True)
+        if directory is None:
+            return
         claim = directory / f"{container_name}.{os.getpid()}"
         try:
-            directory.mkdir(parents=True, exist_ok=True)
             claim.touch()
         except OSError:
             # Unclaimable state directory: reaping then just keeps containers.
@@ -696,15 +730,26 @@ def claimed_containers(cls) -> set:
         found.
         """
         claimed = set()
+        directory = cls._private_directory(create=False)
+        if directory is None:
+            return claimed
         try:
-            claims = list(cls.directory().iterdir())
+            claims = list(directory.iterdir())
         except OSError:
             return claimed
 
+        uid = os.getuid()
         for claim in claims:
             name, _, pid = claim.name.rpartition(".")
             if not pid.isdigit():
                 continue
+            try:
+                info = os.lstat(claim)
+            except OSError:
+                continue
+            if not stat.S_ISREG(info.st_mode) or info.st_uid != uid:
+                # Planted while the directory was open to others.
+                continue
             try:
                 os.kill(int(pid), 0)
             except ProcessLookupError:

tig-cli/tests/test_container.py

@@ -936,6 +936,96 @@ def test_releasing_the_claim_makes_the_container_reapable(home_dir):
     manager.release_claim()  # idempotent
 
 
+def test_the_claim_directory_is_private_to_the_user(home_dir):
+    claim = spec.Claim()
+    claim.acquire("tig-vicar-private")
+
+    info = os.lstat(spec.Claim.directory())
+    assert info.st_mode & 0o777 == 0o700
+    assert info.st_uid == os.getuid()
+    assert claim.path.exists()
+
+
+def test_an_open_claim_directory_of_the_users_own_is_closed(home_dir):
+    directory = spec.Claim.directory()
+    directory.mkdir(parents=True)
+    directory.chmod(0o777)
+
+    claim = spec.Claim()
+    claim.acquire("tig-vicar-open")
+
+    assert os.lstat(directory).st_mode & 0o777 == 0o700
+    assert claim.path is not None
+
+
+def test_a_symlink_in_place_of_the_claim_directory_is_not_trusted(
+    home_dir, tmp_path
+):
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.mkdir()
+    (elsewhere / f"tig-vicar-planted.{os.getpid()}").touch()
+    spec.Claim.directory().symlink_to(elsewhere)
+
+    claim = spec.Claim()
+    claim.acquire("tig-vicar-linked")
+
+    assert claim.path is None
+    assert list(elsewhere.iterdir()) == [
+        elsewhere / f"tig-vicar-planted.{os.getpid()}"
+    ]
+    assert spec.Claim.claimed_containers() == set()
+
+
+def test_a_claim_directory_owned_by_another_user_is_not_trusted(
+    home_dir, monkeypatch
+):
+    other = os.getuid() + 1
+    monkeypatch.setattr(spec.os, "getuid", lambda: other)
+    directory = spec.Claim.directory()
+    directory.mkdir(mode=0o700)
+    (directory / f"tig-vicar-planted.{os.getpid()}").touch()
+
+    claim = spec.Claim()
+    claim.acquire("tig-vicar-foreign")
+
+    assert claim.path is None
+    assert spec.Claim.claimed_containers() == set()
+
+
+def test_claims_not_owned_by_the_user_are_ignored(home_dir, monkeypatch):
+    directory = spec.Claim.directory()
+    directory.mkdir(mode=0o700)
+    (directory / f"tig-vicar-planted.{os.getpid()}").touch()
+    (directory / f"tig-vicar-linked.{os.getpid()}").symlink_to(directory)
+    real_lstat = os.lstat
+
+    def lstat(path):
+        info = real_lstat(path)
+        if "planted" in os.fspath(path):
+            return os.stat_result(
+                info[:4] + (os.getuid() + 1,) + info[5:]
+            )
+        return info
+
+    monkeypatch.setattr(spec.os, "lstat", lstat)
+
+    assert spec.Claim.claimed_containers() == set()
+
+
+def test_claims_fall_back_to_the_temporary_directory(
+    home_dir, tmp_path, monkeypatch
+):
+    import tempfile
+
+    monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False)
+    monkeypatch.setenv("TMPDIR", str(tmp_path))
+    monkeypatch.setattr(tempfile, "tempdir", None)
+
+    assert spec.Claim.directory() == (
+        tmp_path / f"{spec.CLAIM_DIR_NAME}-{os.getuid()}"
+    )
+
+
 def test_reaping_tolerates_a_container_removed_concurrently(home_dir):
     runtime = FakeRuntime({
         f"{CONTAINER_PREFIX}-a": described(),