запуск shell команд под отдельным юзером

Dockerfile

@@ -6,6 +6,11 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
 WORKDIR /app
 RUN apt update && apt install make -y
 
+ENV AGENT_USER="agent"
+RUN useradd --shell /bin/bash agent
+ENV WORKSPACE_DIR="/workspace/"
+RUN mkdir -p $WORKSPACE_DIR && chown $AGENT_USER:$AGENT_USER $WORKSPACE_DIR
+
 FROM base as builder
 
 RUN apt install git -y
@@ -24,12 +29,6 @@ COPY src/ /app/src/
 COPY Makefile ./
 COPY .mk/ ./.mk/
 
-RUN useradd --shell /bin/bash appuser
-USER appuser
-
-ENV WORKSPACE_DIR="/workspace/"
-RUN mkdir -p $WORKSPACE_DIR && chown appuser:appuser $WORKSPACE_DIR
-
 EXPOSE 8000
 
 CMD ["make", "uvicorn-prod"]
@@ -49,8 +48,6 @@ ENV PATH="/app/.venv/bin:$PATH"
 COPY Makefile ./
 COPY .mk/ ./.mk/
 
-ENV WORKSPACE_DIR="/workspace/"
-
 EXPOSE 8000
 
 CMD ["make", "uvicorn-dev"]

src/agent/backends/__init__.py

@@ -0,0 +1,3 @@
+from src.agent.backends.isolated_shell import IsolatedShellBackend
+
+__all__ = ["IsolatedShellBackend"]

src/agent/backends/isolated_shell.py

@@ -0,0 +1,96 @@
+import os
+import pwd
+import subprocess
+from typing import Any
+
+from deepagents.backends.local_shell import LocalShellBackend, DEFAULT_EXECUTE_TIMEOUT
+
+
+class IsolatedShellBackend(LocalShellBackend):
+    """LocalShellBackend с изоляцией shell-команд через отдельного пользователя."""
+
+    def __init__(
+        self,
+        user: str,
+        **kwargs: Any,
+    ):
+        super().__init__(**kwargs)
+        self._user = user
+        self._uid = pwd.getpwnam(user).pw_uid  # type: ignore[attr-defined]
+        self._gid = pwd.getpwnam(user).pw_gid  # type: ignore[attr-defined]
+
+    def execute(
+        self,
+        command: str,
+        *,
+        timeout: int | None = None,
+    ) -> Any:
+        if not command or not isinstance(command, str):
+            return type(self)._error_response("Command must be a non-empty string.")
+
+        effective_timeout = timeout if timeout is not None else self._default_timeout
+        if effective_timeout <= 0:
+            return type(self)._error_response(
+                f"timeout must be positive, got {effective_timeout}"
+            )
+
+        proc: subprocess.Popen[str] | None = None
+        try:
+            proc = subprocess.Popen(
+                command,
+                shell=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                cwd=str(self.cwd),
+                env=self._env,
+                preexec_fn=lambda: (
+                    os.setgid(self._gid) or os.setuid(self._uid)  # type: ignore[attr-defined]
+                ),
+            )
+
+            stdout, stderr = proc.communicate(timeout=effective_timeout)
+
+            output_parts = []
+            if stdout:
+                output_parts.append(stdout)
+            if stderr:
+                stderr_lines = stderr.strip().split("\n")
+                output_parts.extend(f"[stderr] {line}" for line in stderr_lines)
+
+            output = "\n".join(output_parts) if output_parts else "<no output>"
+
+            truncated = False
+            if len(output) > self._max_output_bytes:
+                output = output[: self._max_output_bytes]
+                output += f"\n\n... Output truncated at {self._max_output_bytes} bytes."
+                truncated = True
+
+            if proc.returncode != 0:
+                output = f"{output.rstrip()}\n\nExit code: {proc.returncode}"
+
+            return self._make_response(output, proc.returncode, truncated)
+
+        except subprocess.TimeoutExpired:
+            if proc:
+                proc.kill()
+                proc.communicate()
+            msg = f"Error: Command timed out after {effective_timeout} seconds."
+            return self._make_response(msg, 124, False)
+
+        except Exception as e:
+            return self._make_response(
+                f"Error executing command ({type(e).__name__}): {e}", 1, False
+            )
+
+    @staticmethod
+    def _error_response(message: str):
+        from deepagents.backends.protocol import ExecuteResponse
+
+        return ExecuteResponse(output=message, exit_code=1, truncated=False)
+
+    @staticmethod
+    def _make_response(output: str, exit_code: int, truncated: bool):
+        from deepagents.backends.protocol import ExecuteResponse
+
+        return ExecuteResponse(output=output, exit_code=exit_code, truncated=truncated)

src/agent/base.py

@@ -1,8 +1,10 @@
 import os
+
 from deepagents import create_deep_agent
 from langchain_openai import ChatOpenAI
 from langgraph.checkpoint.memory import MemorySaver
-from deepagents.backends.local_shell import LocalShellBackend
+
+from src.agent.backends import IsolatedShellBackend
 
 
 def create_agent():
@@ -13,8 +15,13 @@ def create_agent():
     )
 
     workspace_dir = os.environ["WORKSPACE_DIR"]
-    backend = LocalShellBackend(workspace_dir,
-                                virtual_mode=True)
+    agent_user = os.environ["AGENT_USER"]
+
+    backend = IsolatedShellBackend(
+        user=agent_user,
+        root_dir=workspace_dir,
+        virtual_mode=True,
+    )
 
     return create_deep_agent(
         model=model,
@@ -22,4 +29,3 @@ def create_agent():
         checkpointer=MemorySaver(),
         backend=backend,
     )
-