fix(hermes_cli): fail-closed PID-ownership guard before Windows taskkill

hermes_cli/_subprocess_compat.py

@@ -46,6 +46,7 @@
     "bounded_git_probe",
     "bounded_probe_run",
     "noninteractive_git_env",
+    "pid_is_hermes",
 ]
 
 
@@ -387,7 +388,70 @@ def noninteractive_git_env(
 # -----------------------------------------------------------------------------
 
 
-def kill_process_tree(proc: "subprocess.Popen") -> None:
+
+def _process_start_time(pid: int) -> int | None:
+    """Return the repository's stable process-start fingerprint, if available."""
+    try:
+        from gateway.status import get_process_start_time
+
+        return get_process_start_time(pid)
+    except Exception:
+        return None
+
+
+def _process_command_is_hermes(pid: int) -> bool:
+    """Best-effort check that *pid* currently runs Hermes code."""
+    try:
+        import psutil
+
+        process = psutil.Process(pid)
+        command = " ".join(process.cmdline() or [])
+        executable = process.exe() or ""
+        return "hermes" in f"{command} {executable}".lower()
+    except Exception:
+        return False
+
+
+def pid_is_hermes(
+    pid: int,
+    *,
+    expected_start_time: int | None = None,
+) -> bool:
+    """Return whether it is safe to use ``taskkill`` for *pid*.
+
+    The PID must be valid, currently exist, and identify a Hermes process. When
+    the caller captured a start-time fingerprint before the destructive action,
+    the live process must still have the same ``(pid, start_time)`` identity.
+    Any ambiguity fails closed. Non-Windows callers have no ``taskkill`` path,
+    so a valid PID is accepted there.
+    """
+    if not isinstance(pid, int) or isinstance(pid, bool) or pid <= 0:
+        return False
+    if not IS_WINDOWS:
+        return True
+
+    try:
+        current_start_time = _process_start_time(pid)
+    except Exception:
+        return False
+    if current_start_time is None:
+        return False
+    if (
+        expected_start_time is not None
+        and current_start_time != expected_start_time
+    ):
+        return False
+    try:
+        return _process_command_is_hermes(pid)
+    except Exception:
+        return False
+
+
+def kill_process_tree(
+    proc: "subprocess.Popen",
+    *,
+    expected_start_time: int | None = None,
+) -> None:
     """Best-effort terminate *proc* and its descendants on both platforms.
 
     ``proc.kill()`` alone only terminates the direct child. On Windows a
@@ -430,8 +494,19 @@ def kill_process_tree(proc: "subprocess.Popen") -> None:
         pass
     if IS_WINDOWS:
         try:
-            subprocess.run(
-                ["taskkill", "/T", "/F", "/PID", str(proc.pid)],
+            live_start_time = expected_start_time
+            if live_start_time is None:
+                live_start_time = _process_start_time(proc.pid)
+            if live_start_time is None:
+                allowed = False
+            else:
+                allowed = pid_is_hermes(
+                    proc.pid,
+                    expected_start_time=live_start_time,
+                )
+            if allowed:
+                subprocess.run(
+                    ["taskkill", "/T", "/F", "/PID", str(proc.pid)],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 stdin=subprocess.DEVNULL,
@@ -495,7 +570,10 @@ def bounded_probe_run(
         # Timeout OR any other communicate() failure (torn-down pipe, decode
         # error): terminate the child + descendants and drain bounded. Leaving
         # it running would leak the same suspended-descendant class this guards.
-        kill_process_tree(proc)
+        kill_process_tree(
+            proc,
+            expected_start_time=_process_start_time(proc.pid),
+        )
         try:
             proc.communicate(timeout=1)
         except Exception:

hermes_cli/dashboard_procs.py

@@ -400,13 +400,37 @@ def _kill_stale_dashboard_processes(
     failed: list[tuple[int, str]] = []
 
     if sys.platform == "win32":
+        from gateway.status import get_process_start_time
+        from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
+
+        # Capture the identity immediately after discovery. A PID that is
+        # reused before the destructive action will fail the start-time check.
+        pid_start_times = {
+            pid: get_process_start_time(pid)
+            for pid in pids
+        }
         for pid in pids:
             try:
+                expected_start_time = pid_start_times.get(pid)
+                if expected_start_time is None:
+                    failed.append((pid, "could not verify process identity"))
+                    continue
+                if not pid_is_hermes(
+                    pid,
+                    expected_start_time=expected_start_time,
+                ):
+                    failed.append((pid, "not hermes-owned or process identity changed"))
+                    continue
                 result = subprocess.run(
                     ["taskkill", "/PID", str(pid), "/F"],
-                    capture_output=True,
-                    text=True, encoding="utf-8", errors="replace",
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    stdin=subprocess.DEVNULL,
+                    text=True,
+                    encoding="utf-8",
+                    errors="replace",
                     timeout=10,
+                    creationflags=windows_hide_flags(),
                 )
                 if result.returncode == 0:
                     killed.append(pid)

hermes_cli/update_cmd.py

@@ -4504,7 +4504,7 @@ def _leftover_pausable_gateway_pids(
 
 def _orphaned_desktop_backend_pids(
     matches: list[tuple[int, str, str]],
-) -> list[int] | None:
+) -> list[tuple[int, int]] | None:
     """PIDs from *matches* when every remaining holder is an ORPHANED backend.
 
     The venv-holder guard refuses on the Desktop app's ``serve`` backend by
@@ -4552,7 +4552,7 @@ def _is_backend(argv_low: str) -> bool:
         )
 
     # Pass 1: find orphaned backend ROOTS among the holders.
-    roots: list[int] = []
+    roots: list[tuple[int, int]] = []
     remaining: list[tuple[int, str]] = []  # (pid, argv_low) still to justify
     for pid, _name, cmdline in matches:
         argv = cmdline
@@ -4570,6 +4570,19 @@ def _is_backend(argv_low: str) -> bool:
             continue
         try:
             proc = psutil.Process(int(pid))
+            from gateway.status import get_process_start_time
+
+            process_start_time = get_process_start_time(int(pid))
+            if process_start_time is None:
+                return None
+        except psutil.NoSuchProcess:
+            # The candidate itself exited during classification; there is
+            # nothing left to reap and no identity to pass to taskkill.
+            continue
+        except Exception:
+            return None
+
+        try:
             ppid = proc.ppid()
             parent = psutil.Process(ppid) if ppid else None
             if parent is not None and parent.is_running():
@@ -4588,12 +4601,12 @@ def _is_backend(argv_low: str) -> bool:
             pass  # parent gone → orphan
         except Exception:
             return None
-        roots.append(int(pid))
+        roots.append((int(pid), process_start_time))
 
     # Pass 2: every non-backend holder must be a descendant of an accepted
     # orphan root — then it dies with the root's tree reap. Anything else
     # (operator REPL, stray script) keeps the refusal.
-    root_set = set(roots)
+    root_set = {pid for pid, _start_time in roots}
     for pid, _low in remaining:
         if not root_set:
             return None
@@ -4723,20 +4736,48 @@ def _is_backend(argv_low: str) -> bool:
     return roots or None
 
 
-def _stop_process_trees(pids: list[int]) -> None:
+def _stop_process_trees(
+    pids: list[int] | list[tuple[int, int]],
+) -> None:
     """Force-stop each PID with its full child tree (Windows).
 
     ``taskkill /T /F`` mirrors the Desktop's ``forceKillProcessTree`` and
     install.ps1's venv sweep: stopping only the parent can leave a managed
     ``.hermes-runtime`` interpreter child alive and holding the install open
     (#70026). Best effort; never raises.
     """
-    for pid in pids:
+    from gateway.status import get_process_start_time
+    from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
+
+    for entry in pids:
+        if isinstance(entry, tuple):
+            pid, expected_start_time = entry
+        else:
+            pid = int(entry)
+            expected_start_time = get_process_start_time(pid)
         try:
+            if expected_start_time is None:
+                logger.debug(
+                    "Skipping taskkill of PID %s: process identity unavailable",
+                    pid,
+                )
+                continue
+            if not pid_is_hermes(
+                pid,
+                expected_start_time=expected_start_time,
+            ):
+                logger.debug(
+                    "Skipping taskkill of non-Hermes or changed PID %s",
+                    pid,
+                )
+                continue
             subprocess.run(
-                ["taskkill", "/PID", str(int(pid)), "/T", "/F"],
+                ["taskkill", "/PID", str(pid), "/T", "/F"],
                 check=False,
-                capture_output=True,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                stdin=subprocess.DEVNULL,
+                creationflags=windows_hide_flags(),
             )
         except Exception as exc:
             logger.debug("Could not stop process tree %s: %s", pid, exc)