Skip sudo wrapper when os_user matches bot process user

src/kai/claude.py

@@ -29,6 +29,7 @@
 import json
 import logging
 import os
+import pwd
 import signal
 import time
 from collections.abc import AsyncIterator
@@ -210,15 +211,33 @@ async def _ensure_started(self) -> None:
         # When running as a different user, spawn via sudo -u.
         # The subprocess runs with the target user's UID, home directory,
         # and environment - completely isolated from the bot user.
-        if self.claude_user:
-            cmd = ["sudo", "-u", self.claude_user, "--"] + claude_cmd
+        # Skip sudo when the target user is the same as the bot user -
+        # self-sudo fails (sudoers disallows it) and serves no purpose.
+        effective_claude_user = self.claude_user
+        if effective_claude_user:
+            # getpwuid raises KeyError when the UID has no passwd entry
+            # (e.g., containers with --user <uid>). Treat that as "unknown
+            # user" and fall through to the sudo path.
+            try:
+                current_user = pwd.getpwuid(os.getuid()).pw_name
+            except KeyError:
+                current_user = None
+            if effective_claude_user == current_user:
+                log.warning(
+                    "claude_user %r matches the bot process user; skipping sudo (no isolation)",
+                    effective_claude_user,
+                )
+                effective_claude_user = None
+
+        if effective_claude_user:
+            cmd = ["sudo", "-u", effective_claude_user, "--"] + claude_cmd
         else:
             cmd = claude_cmd
 
         log.info(
             "Starting persistent Claude process (model=%s, user=%s)",
             self.model,
-            self.claude_user or "(same as bot)",
+            effective_claude_user or "(same as bot)",
         )
 
         # Build the subprocess environment. Merge order:
@@ -247,7 +266,7 @@ async def _ensure_started(self) -> None:
             # When spawned via sudo, start in a new process group so we can
             # kill the entire tree (sudo + claude) via os.killpg(). Without
             # this, killing sudo may orphan the claude process.
-            start_new_session=bool(self.claude_user),
+            start_new_session=bool(effective_claude_user),
         )
         self._session_id = None
         self._fresh_session = True
@@ -258,7 +277,7 @@ async def _ensure_started(self) -> None:
         # with PGID == PID (session leader). Save it now because os.getpgid()
         # fails after the process exits, but os.killpg() works as long as any
         # group member is still alive (i.e., the actual claude process).
-        if self.claude_user:
+        if effective_claude_user:
             self._pgid = self._proc.pid  # PGID == PID for session leaders
         else:
             self._pgid = None

tests/test_claude.py

@@ -14,6 +14,8 @@
 
 import asyncio
 import json
+import os
+import pwd
 import signal
 import time
 from pathlib import Path
@@ -171,6 +173,84 @@ async def test_start_new_session_true_with_claude_user(self):
             args = mock_exec.call_args
             assert args[1].get("start_new_session") is True
 
+    @pytest.mark.asyncio
+    async def test_self_sudo_skipped(self, caplog):
+        """When claude_user matches the current process user, sudo is skipped."""
+        try:
+            current_user = pwd.getpwuid(os.getuid()).pw_name
+        except KeyError:
+            pytest.skip("UID has no passwd entry; self-sudo-skip path not reachable")
+        claude = _make_claude(claude_user=current_user)
+
+        with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec:
+            mock_proc = MagicMock()
+            mock_proc.returncode = None
+            mock_proc.stderr = AsyncMock()
+            mock_exec.return_value = mock_proc
+
+            with caplog.at_level("WARNING", logger="kai.claude"):
+                await claude._ensure_started()
+
+            args = mock_exec.call_args
+            cmd = args[0]
+            # Command should NOT start with sudo
+            assert cmd[0] == "claude"
+            assert "sudo" not in cmd
+            # No process group isolation
+            assert args[1].get("start_new_session") is False
+            assert claude._pgid is None
+            # Warning was logged
+            assert "skipping sudo" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_different_user_still_uses_sudo(self):
+        """When claude_user is a different user, sudo is still used."""
+        claude = _make_claude(claude_user="some_other_user")
+
+        # Patch pwd.getpwuid to return a fixed value so the test doesn't
+        # depend on the real system user being different from "some_other_user".
+        mock_pw = MagicMock(pw_name="kai")
+        with (
+            patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
+            patch("kai.claude.pwd.getpwuid", return_value=mock_pw),
+        ):
+            mock_proc = MagicMock()
+            mock_proc.returncode = None
+            mock_proc.stderr = AsyncMock()
+            mock_exec.return_value = mock_proc
+
+            await claude._ensure_started()
+
+            args = mock_exec.call_args
+            cmd = args[0]
+            assert cmd[0] == "sudo"
+            assert cmd[2] == "some_other_user"
+            assert args[1].get("start_new_session") is True
+
+    @pytest.mark.asyncio
+    async def test_pwd_keyerror_falls_through_to_sudo(self):
+        """When pwd.getpwuid raises KeyError (unmapped UID), sudo is used."""
+        claude = _make_claude(claude_user="container_user")
+
+        # Simulate a container environment where the UID has no passwd entry.
+        with (
+            patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
+            patch("kai.claude.pwd.getpwuid", side_effect=KeyError("uid not found")),
+        ):
+            mock_proc = MagicMock()
+            mock_proc.returncode = None
+            mock_proc.stderr = AsyncMock()
+            mock_exec.return_value = mock_proc
+
+            await claude._ensure_started()
+
+            args = mock_exec.call_args
+            cmd = args[0]
+            # Should still wrap with sudo since we can't determine the user.
+            assert cmd[0] == "sudo"
+            assert cmd[2] == "container_user"
+            assert args[1].get("start_new_session") is True
+
 
 # ── Process signal handling ──────────────────────────────────────────