Keep capacity worker pid files private and resolve Git without PATH (#1051)

scripts/benchmark/capacity_local_docker.py

@@ -633,7 +633,11 @@ def _execute_matrix(
         topology, topology["component_services"]["load-generator"]
     )
     environment = _execution_environment(context)
-    source_revision = _run(["git", "rev-parse", "HEAD"], timeout=30).strip()
+    try:
+        git = capacity_suite.git_executable()
+    except FileNotFoundError as error:
+        raise TopologyError(str(error)) from error
+    source_revision = _run([git, "rev-parse", "HEAD"], timeout=30).strip()
     if not re.fullmatch(r"[0-9a-f]{40}", source_revision):
         raise TopologyError("source revision must be a full Git commit")
     matrix_arguments = [command, *arguments]

scripts/benchmark/capacity_matrix.py

@@ -38,6 +38,44 @@
 DEFAULT_COLLECTOR = SUITE_ROOT / "collectors/local-docker/collector.json"
 DEFAULT_SMOKE = SUITE_ROOT / "topologies/local-docker/smoke.json"
 
+# Container worker lifecycle files (Server #1051). Each script takes only the
+# file name and keeps it in one private directory under the container's temp
+# directory: created 0700, and used only when it is a real directory owned by
+# the adapter user, so another user cannot plant or redirect a pid file. Each
+# file is created exclusively (noclobber), so a stale or planted file refuses
+# the start instead of being overwritten. Cleanup checks the same ownership
+# and still kills only the process whose recorded start time matches.
+_PIDFILE_CUSTODY = (
+    'case "$1" in ""|.*|*/*) exit 42;; esac; '
+    'dir="${TMPDIR:-/tmp}/durable-workflow-capacity-workers"; pidfile="$dir/$1"; '
+)
+_PIDFILE_DIRECTORY_OWNED = (
+    '{ [ -d "$dir" ] && [ ! -L "$dir" ] && [ -O "$dir" ]; } || exit 42; '
+)
+WORKER_PIDFILE_SCRIPT = (
+    _PIDFILE_CUSTODY
+    + 'shift; umask 077; mkdir -m 700 "$dir" 2>/dev/null; '
+    + _PIDFILE_DIRECTORY_OWNED
+    + 'start="$(awk \'{print $22}\' "/proc/$$/stat")"; '
+    'set -C; printf "%s %s\\n" "$$" "$start" >"$pidfile" || exit 43; set +C; '
+    'exec "$@"'
+)
+WORKER_CLEANUP_SCRIPT = (
+    _PIDFILE_CUSTODY
+    + 'if [ ! -e "$pidfile" ] && [ ! -L "$pidfile" ]; then exit 0; fi; '
+    + _PIDFILE_DIRECTORY_OWNED
+    + '{ [ -f "$pidfile" ] && [ ! -L "$pidfile" ] && [ -O "$pidfile" ]; } || exit 42; '
+    'read -r pid expected_start <"$pidfile"; '
+    'if [ -r "/proc/$pid/stat" ]; then '
+    'actual_start="$(awk \'{print $22}\' "/proc/$pid/stat")"; '
+    '[ "$actual_start" = "$expected_start" ] || exit 41; '
+    'kill -TERM "$pid"; attempts=0; '
+    'while kill -0 "$pid" 2>/dev/null && [ "$attempts" -lt 150 ]; do '
+    "sleep 0.1; attempts=$((attempts + 1)); done; "
+    'if kill -0 "$pid" 2>/dev/null; then kill -KILL "$pid"; fi; '
+    'fi; rm -f "$pidfile"'
+)
+
 
 class MatrixError(RuntimeError):
     """The executable benchmark matrix failed closed."""
@@ -833,11 +871,7 @@ def _entrypoint(
                 [
                     "sh",
                     "-c",
-                    (
-                        'pidfile="$1"; shift; umask 077; '
-                        'start="$(awk \'{print $22}\' "/proc/$$/stat")"; '
-                        'printf "%s %s\\n" "$$" "$start" >"$pidfile"; exec "$@"'
-                    ),
+                    WORKER_PIDFILE_SCRIPT,
                     "capacity-worker",
                     self._container_worker_pid_file(descriptor, environment),
                     *command,
@@ -860,7 +894,7 @@ def _container_worker_pid_file(
             )
         )
         digest = hashlib.sha256(identity.encode()).hexdigest()[:24]
-        return f"/tmp/durable-workflow-capacity-worker-{digest}.pid"
+        return f"{digest}.pid"
 
     def _container_worker_cleanup_command(
         self,
@@ -875,26 +909,13 @@ def _container_worker_cleanup_command(
         if not container:
             return None
         pidfile = self._container_worker_pid_file(descriptor, environment)
-        script = (
-            'pidfile="$1"; '
-            'if [ ! -f "$pidfile" ]; then exit 0; fi; '
-            'read -r pid expected_start <"$pidfile"; '
-            'if [ -r "/proc/$pid/stat" ]; then '
-            'actual_start="$(awk \'{print $22}\' "/proc/$pid/stat")"; '
-            '[ "$actual_start" = "$expected_start" ] || exit 41; '
-            'kill -TERM "$pid"; attempts=0; '
-            'while kill -0 "$pid" 2>/dev/null && [ "$attempts" -lt 150 ]; do '
-            "sleep 0.1; attempts=$((attempts + 1)); done; "
-            'if kill -0 "$pid" 2>/dev/null; then kill -KILL "$pid"; fi; '
-            'fi; rm -f "$pidfile"'
-        )
         return [
             "docker",
             "exec",
             container,
             "sh",
             "-c",
-            script,
+            WORKER_CLEANUP_SCRIPT,
             "capacity-worker-cleanup",
             pidfile,
         ]

scripts/benchmark/capacity_suite.py

@@ -7,6 +7,7 @@
 import hashlib
 import json
 import math
+import os
 import platform
 import re
 import subprocess
@@ -2589,10 +2590,31 @@ def _reference_observations(
     return observations
 
 
+# System Git installs, checked in order. The capacity tools never resolve Git
+# through PATH (Server #1051); CAPACITY_GIT names another absolute binary.
+GIT_CANDIDATES = ("/usr/bin/git", "/usr/local/bin/git", "/opt/homebrew/bin/git")
+
+
+def git_executable() -> str:
+    configured = os.environ.get("CAPACITY_GIT", "").strip()
+    for candidate in (configured,) if configured else GIT_CANDIDATES:
+        path = Path(candidate)
+        if path.is_absolute() and path.is_file() and os.access(path, os.X_OK):
+            return str(path)
+    raise FileNotFoundError(
+        "CAPACITY_GIT must name an absolute, executable Git binary"
+        if configured
+        else "no Git executable at " + ", ".join(GIT_CANDIDATES)
+    )
+
+
 def _git_revision() -> str:
     try:
         return subprocess.check_output(
-            ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, stderr=subprocess.DEVNULL
+            [git_executable(), "rev-parse", "HEAD"],
+            cwd=ROOT,
+            text=True,
+            stderr=subprocess.DEVNULL,
         ).strip()
     except (OSError, subprocess.CalledProcessError):
         return "unknown-local-revision"

tests/Unit/Support/capacity_benchmark_test.py

@@ -4,6 +4,7 @@
 import importlib.util
 import io
 import json
+import os
 from pathlib import Path
 import shutil
 import subprocess
@@ -643,12 +644,10 @@ def test_matrix_routes_adapter_roles_to_resolved_language_containers(
         self.assertEqual(["python3", "capacity_adapter.py", "worker"], command[-3:])
         self.assertNotIn("--interactive", command)
         self.assertIn("capacity-worker", command)
-        self.assertTrue(
-            any(
-                value.startswith("/tmp/durable-workflow-capacity-worker-")
-                for value in command
-            )
+        self.assertRegex(
+            runner._container_worker_pid_file(adapter, environment), r"^[0-9a-f]{24}\.pid$"
         )
+        self.assertIn(runner._container_worker_pid_file(adapter, environment), command)
         self.assertTrue(
             any(
                 value == "DURABLE_WORKFLOW_TASK_QUEUE=capacity-smoke"
@@ -679,6 +678,84 @@ def test_matrix_routes_adapter_roles_to_resolved_language_containers(
             cleanup[-1],
         )
 
+    def test_worker_pid_files_are_private_and_exclusive(self) -> None:
+        # Server #1051: run the exact container scripts with the local sh,
+        # with TMPDIR standing in for the container's temp directory.
+        def run(script: str, temp: Path, *arguments: str) -> subprocess.CompletedProcess[str]:
+            return subprocess.run(
+                ["sh", "-c", script, "capacity-worker", *arguments],
+                text=True, capture_output=True, check=False,
+                env={"PATH": os.environ.get("PATH", ""), "TMPDIR": str(temp)},
+            )
+
+        def worker(temp: Path, name: str) -> subprocess.CompletedProcess[str]:
+            return run(capacity_matrix.WORKER_PIDFILE_SCRIPT, temp, name, "true")
+
+        def cleanup(temp: Path, name: str) -> subprocess.CompletedProcess[str]:
+            return run(capacity_matrix.WORKER_CLEANUP_SCRIPT, temp, name)
+
+        with tempfile.TemporaryDirectory() as scratch:
+            temp = Path(scratch) / "temp"
+            temp.mkdir()
+            directory = temp / "durable-workflow-capacity-workers"
+            pidfile = directory / "worker.pid"
+            started = worker(temp, "worker.pid")
+            self.assertEqual(0, started.returncode, started.stderr)
+            self.assertEqual(0o700, directory.stat().st_mode & 0o777)
+            self.assertEqual(0o600, pidfile.stat().st_mode & 0o777)
+            recorded = pidfile.read_text()
+            self.assertRegex(recorded, r"^\d+ \d+\n$")
+
+            # A second start with the same identity refuses; nothing is overwritten.
+            self.assertEqual(43, worker(temp, "worker.pid").returncode)
+            self.assertEqual(recorded, pidfile.read_text())
+
+            # The recorded process has exited, so cleanup only removes the file.
+            self.assertEqual(0, cleanup(temp, "worker.pid").returncode)
+            self.assertFalse(pidfile.exists())
+            self.assertEqual(0, cleanup(temp, "worker.pid").returncode, "a missing pid file is already clean")
+
+            # Only a plain file name is accepted.
+            for name in ("", "../escape.pid", ".hidden", "a/b.pid"):
+                self.assertEqual(42, worker(temp, name).returncode, name)
+
+            # A symlinked directory or pid file is refused, never followed.
+            target = Path(scratch) / "elsewhere"
+            target.mkdir(mode=0o700)
+            linked = Path(scratch) / "linked-temp"
+            linked.mkdir()
+            (linked / "durable-workflow-capacity-workers").symlink_to(target)
+            self.assertEqual(42, worker(linked, "worker.pid").returncode)
+            self.assertEqual([], list(target.iterdir()))
+            (directory / "planted.pid").symlink_to(target / "victim")
+            self.assertEqual(42, cleanup(temp, "planted.pid").returncode)
+            self.assertFalse((target / "victim").exists())
+
+            # A live process with a different recorded start is left alone.
+            with open(pidfile, "x") as handle:
+                handle.write(f"{os.getpid()} 1\n")
+            self.assertEqual(41, cleanup(temp, "worker.pid").returncode)
+            self.assertTrue(pidfile.exists())
+
+    def test_git_is_resolved_without_path(self) -> None:
+        with tempfile.TemporaryDirectory() as scratch:
+            fake = Path(scratch) / "git"
+            fake.write_text("#!/bin/sh\necho 0123456789abcdef0123456789abcdef01234567\n")
+            fake.chmod(0o700)
+            with mock.patch.dict(capacity_suite.os.environ, {"CAPACITY_GIT": str(fake), "PATH": scratch}):
+                self.assertEqual(str(fake), capacity_suite.git_executable())
+                self.assertEqual("0123456789abcdef0123456789abcdef01234567", capacity_suite._git_revision())
+            for configured in ("git", str(Path(scratch) / "missing")):
+                with mock.patch.dict(capacity_suite.os.environ, {"CAPACITY_GIT": configured}):
+                    with self.assertRaises(FileNotFoundError):
+                        capacity_suite.git_executable()
+                    self.assertEqual("unknown-local-revision", capacity_suite._git_revision())
+            # Unset, a PATH entry named git is never chosen.
+            with mock.patch.dict(capacity_suite.os.environ, {"PATH": scratch}, clear=True):
+                with mock.patch.object(capacity_suite, "GIT_CANDIDATES", ("/nonexistent/git",)):
+                    with self.assertRaises(FileNotFoundError):
+                        capacity_suite.git_executable()
+
     def test_worker_process_runs_container_cleanup_after_client_exit(self) -> None:
         process = mock.Mock()
         process.poll.return_value = 0