Remove sudo access to agent shell

src/microbots/bot/CopilotBot.py

@@ -535,7 +535,7 @@ def _install_copilot_cli(self):
         ]
 
         for cmd in install_commands:
-            result = self.environment.execute(cmd, timeout=300)
+            result = self.environment.execute_privileged(cmd, timeout=300)
             if result.return_code != 0:
                 raise RuntimeError(
                     f"Failed to install copilot-cli: {cmd}\n"
@@ -573,7 +573,7 @@ def _start_copilot_cli_server(self):
         # Using nohup + & to run it as a background process inside the container's shell
         start_cmd = (
             f"nohup copilot --headless --port {_CONTAINER_CLI_PORT} --host 0.0.0.0 "
-            f"> /var/log/copilot-cli.log 2>&1 &"
+            f"> /var/log/microbots/copilot-cli.log 2>&1 &"
         )
         result = self.environment.execute(start_cmd)
         if result.return_code != 0:
@@ -604,7 +604,7 @@ def _wait_for_cli_ready(self):
                 return
             except (ConnectionRefusedError, OSError):
                 time.sleep(1)
-        self.environment.execute("cat /var/log/copilot-cli.log || true")
+        self.environment.execute("cat /var/log/microbots/copilot-cli.log || true")
         raise TimeoutError(
             f"copilot-cli did not become ready within {_CLI_STARTUP_TIMEOUT}s "
             f"on {container_ip}:{_CONTAINER_CLI_PORT}"

src/microbots/environment/Environment.py

@@ -22,6 +22,12 @@ def stop(self):
     def execute(self, command: str, timeout: Optional[int] = 300, sensitive: bool = False) -> CmdReturn:
         pass
 
+    def execute_privileged(self, command: str, timeout: Optional[int] = 300, sensitive: bool = False) -> CmdReturn:
+        """Run a command on the control plane, which may hold more privilege
+        than the bot's own channel. Defaults to the bot channel.
+        Override this one while implementing custom Environment subclasses."""
+        return self.execute(command, timeout=timeout, sensitive=sensitive)
+
     def copy_to_container(self, src_path: str, dest_path: str) -> bool:
         raise NotImplementedError(
             f"{self.__class__.__name__} does not support copying files to container. "

src/microbots/environment/local_docker/LocalDockerEnvironment.py

@@ -79,17 +79,25 @@ def start(self):
                     self.folder_to_mount.sandbox_path,
                 )
 
-        # Port mapping
-        port_mapping = {f"{self.container_port}/tcp": self.port}
+        # Bind to loopback only: the shell server has no authentication, so it
+        # must never be reachable from the network.
+        port_mapping = {f"{self.container_port}/tcp": ("127.0.0.1", self.port)}
 
         self.container = self.client.containers.run(
             self.image,
             volumes=volumes_config,
             ports=port_mapping,
             detach=True,
             working_dir="/app",
-            privileged=True,  # Required for mounting overlayfs
-            environment={"BOT_PORT": str(self.container_port)},
+            # SYS_ADMIN is the narrowest grant that allows the overlayfs mount;
+            # the docker-default AppArmor profile denies mount regardless of caps.
+            cap_add=["SYS_ADMIN"],
+            security_opt=["no-new-privileges:true", "apparmor=unconfined"],
+            environment={
+                "BOT_PORT": str(self.container_port),
+                "BOT_WORKDIR": DOCKER_WORKING_DIR,
+                **self._host_identity(),
+            },
         )
         logger.info(
             "🚀 Started container %s with image %s on host port %s",
@@ -107,51 +115,80 @@ def start(self):
         else:
             self.execute("cd /")
 
+    @staticmethod
+    def _host_identity() -> dict:
+        """Run the bot's shell under the host uid so anything it writes to the
+        bind-mounted working directory stays removable by the host user."""
+        if not hasattr(os, "getuid") or os.getuid() == 0:
+            return {}
+        return {"AGENT_UID": str(os.getuid()), "AGENT_GID": str(os.getgid())}
+
+    @property
+    def _overlay_dir(self) -> str:
+        path_name = os.path.basename(self.folder_to_mount.sandbox_path)
+        return f"{DOCKER_WORKING_DIR}/overlay/{path_name}"
+
     def _setup_overlay_mount(self):
-        # NOTE: Don't use this for any other read-only mounts except the main code folder.
+        """Stack a writable layer over the READ_ONLY mount so the bot can edit
+        without the host's tree ever changing.
 
+        Runs on the root control channel: the bot's own shell holds no
+        capabilities and cannot mount or unmount anything.
+        """
+        # NOTE: Don't use this for any other read-only mounts except the main code folder.
         path_name = os.path.basename(self.folder_to_mount.sandbox_path)
-        # Mount /ro/path_name to /{WORKING_DIR}/path_name using overlayfs
-        mount_command = (
-            f"mkdir -p {self.folder_to_mount.sandbox_path} /{DOCKER_WORKING_DIR}/overlay/{path_name}/upper /{DOCKER_WORKING_DIR}/overlay/{path_name}/work && sleep 5 && "
-            f"mount -t overlay overlay -o lowerdir=/ro/{path_name}/,upperdir={DOCKER_WORKING_DIR}/overlay/{path_name}/upper/,workdir={DOCKER_WORKING_DIR}/overlay/{path_name}/work/ {self.folder_to_mount.sandbox_path}"
+        sandbox_path = shlex.quote(self.folder_to_mount.sandbox_path)
+        overlay = shlex.quote(self._overlay_dir)
+
+        ret: CmdReturn = self.execute_privileged(
+            f"mkdir -p {sandbox_path} {overlay}/upper {overlay}/work && "
+            f"mount -t overlay overlay "
+            f"-o lowerdir=/ro/{path_name}/,upperdir={overlay}/upper/,workdir={overlay}/work/ "
+            f"{sandbox_path}"
         )
-        self.execute(mount_command)
+        if ret.return_code != 0:
+            raise RuntimeError(
+                f"Failed to set up overlay mount for {path_name}: {ret.stderr}"
+            )
+        self.overlay_mount = True
+
+        # The merged root inherits the lower directory's owner, which may not be
+        # the bot, leaving it unable to create files at the top level.
+        identity = self._host_identity()
+        if identity:
+            self.execute_privileged(
+                f"chown {identity['AGENT_UID']}:{identity['AGENT_GID']} {sandbox_path}"
+            )
+
         logger.info(
-            f"🔒 Set up overlay mount for read-only directory at {DOCKER_WORKING_DIR}/{path_name}"
+            "🔒 Set up overlay mount for read-only directory at %s",
+            self.folder_to_mount.sandbox_path,
         )
-        self.overlay_mount = True
 
     def _teardown_overlay_mount(self):
-        path_name = os.path.basename(os.path.abspath(self.folder_to_mount.sandbox_path))
+        """Unmount and remove the overlay before the container goes away.
 
+        The kernel creates ``work/`` root-owned and mode 0700, so the host user
+        cannot clean it up afterwards - it has to go through the root channel.
+        """
+        sandbox_path = shlex.quote(self.folder_to_mount.sandbox_path)
+        overlay = shlex.quote(self._overlay_dir)
         try:
-            logger.info("🛠️  Tearing down overlay mount for %s", path_name)
-            unmount_command = f"umount -l {self.folder_to_mount.sandbox_path}"
-            ret: CmdReturn = self.execute(unmount_command)
+            ret: CmdReturn = self.execute_privileged(f"umount -l {sandbox_path}")
             if ret.return_code != 0:
                 logger.error("❌  Failed to unmount overlay: %s", ret.stderr)
             else:
-                logger.info("✅  Unmounted overlay for %s", path_name)
+                logger.info("✅  Unmounted overlay at %s", self.folder_to_mount.sandbox_path)
 
-            logger.info(
-                f"🛑  Removing overlay dirs at {self.folder_to_mount.sandbox_path} and {DOCKER_WORKING_DIR}/overlay/"
-            )
-            remove_dir_command = (
-                f"rm -rf {self.folder_to_mount.sandbox_path} && "
-                f"rm -rf {DOCKER_WORKING_DIR}/overlay/"
-            )
-            ret: CmdReturn = self.execute(remove_dir_command)
+            ret = self.execute_privileged(f"rm -rf {sandbox_path} {overlay}")
             if ret.return_code != 0:
-                logger.error(
-                    "❌  Failed to remove overlay directories: %s", ret.stderr
-                )
+                logger.error("❌  Failed to remove overlay directories: %s", ret.stderr)
             else:
-                logger.info(
-                    "🗑️  Removed overlay directories for %s", path_name
-                )
+                logger.info("🗑️  Removed overlay directories for %s", self._overlay_dir)
         except Exception as e:
             logger.error("❌  Failed to teardown overlay mount: %s", e)
+        finally:
+            self.overlay_mount = False
 
     def get_ipv4_address(self) -> str:
         """Return the container's IPv4 address on the Docker bridge network."""
@@ -194,6 +231,31 @@ def _escape(self, command: str) -> str:
         command = command.replace("<", "&lt;").replace(">", "&gt;")
         return command
 
+    def execute_privileged(
+        self, command: str, timeout: Optional[int] = 300, sensitive: bool = False
+    ) -> CmdReturn:
+        """Run a command as root over the docker exec control plane.
+
+        Reserved for setup the bot itself must not perform (package installs,
+        writes outside the working directory). Unlike execute(), this path is
+        not reachable from the container's published port.
+
+        ``timeout`` is accepted for signature parity but not enforced: the
+        docker exec API has no timeout, so a wedged command blocks here.
+        """
+        if not self.container:
+            raise RuntimeError("No active container to execute a privileged command in")
+
+        logger.debug("➡️  Executing privileged command: %s", "<redacted>" if sensitive else command)
+        exit_code, (stdout, stderr) = self.container.exec_run(
+            ["bash", "-lc", command], user="root", demux=True
+        )
+        return CmdReturn(
+            stdout=stdout.decode(errors="replace") if stdout else "",
+            stderr=stderr.decode(errors="replace") if stderr else "",
+            return_code=exit_code if exit_code is not None else 0,
+        )
+
     def execute(
         self, command: str, timeout: Optional[int] = 300, sensitive: bool = False
     ) -> CmdReturn:  # TODO: Need proper return value

src/microbots/environment/local_docker/image_builder/Dockerfile

@@ -10,7 +10,17 @@ RUN pip install --no-cache-dir fastapi uvicorn pydantic
 # command.
 RUN git config --system --add safe.directory '*'
 
-# Copy application files
+# The bot shell runs as this unprivileged account. The devcontainer base image
+# ships a 'vscode' user with NOPASSWD sudo, which would hand root straight back.
+RUN (userdel -r vscode 2>/dev/null || true) && \
+    rm -f /etc/sudoers.d/vscode && \
+    groupadd -g 1001 agent && \
+    useradd -m -u 1001 -g 1001 -s /bin/bash agent && \
+    mkdir -p /var/log/microbots && \
+    chown agent:agent /var/log/microbots
+
+# Copy application files. These stay root-owned so the bot cannot patch the
+# shell server it talks to.
 COPY src/microbots/environment/local_docker/image_builder/dockerShell.py .
 COPY src/microbots/environment/local_docker/image_builder/ShellCommunicator.py .
 

src/microbots/environment/local_docker/image_builder/ShellCommunicator.py

@@ -17,7 +17,7 @@
 from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)
-logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename='/var/log/ShellCommunicator.log')
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename='/var/log/microbots/ShellCommunicator.log')
 
 @dataclass
 class CmdReturn:

src/microbots/environment/local_docker/image_builder/dockerShell.py

@@ -1,5 +1,7 @@
 import os
 import logging
+import subprocess
+from typing import Tuple
 
 import uvicorn
 from fastapi import FastAPI
@@ -19,6 +21,44 @@
 logging.getLogger('ShellCommunicator').setLevel(logging.DEBUG)
 logging.getLogger('uvicorn').setLevel(logging.INFO)
 
+logger = logging.getLogger(__name__)
+
+# Must match the account created in the Dockerfile.
+AGENT_USER = "agent"
+DEFAULT_AGENT_UID = 1001
+DEFAULT_AGENT_GID = 1001
+AGENT_OWNED_PATHS = ("/home/agent", "/var/log/microbots")
+
+
+def _align_agent_with_host_user() -> Tuple[int, int]:
+    """Re-number the agent account to the host uid so files it writes to the
+    bind-mounted working directory stay removable by the host user."""
+    uid = int(os.getenv("AGENT_UID") or DEFAULT_AGENT_UID)
+    gid = int(os.getenv("AGENT_GID") or DEFAULT_AGENT_GID)
+
+    if (uid, gid) != (DEFAULT_AGENT_UID, DEFAULT_AGENT_GID):
+        subprocess.run(["groupmod", "-g", str(gid), AGENT_USER], check=True)
+        subprocess.run(["usermod", "-u", str(uid), "-g", str(gid), AGENT_USER], check=True)
+
+    workdir = os.getenv("BOT_WORKDIR")
+    for path in AGENT_OWNED_PATHS + ((workdir,) if workdir else ()):
+        subprocess.run(["chown", "-R", f"{uid}:{gid}", path], check=False)
+    return uid, gid
+
+
+def _drop_privileges(uid: int, gid: int) -> None:
+    """Irreversibly drop this process (and therefore the bot's shell) to the
+    agent account. Root work happens over the docker exec control channel."""
+    os.setgroups([gid])
+    os.setgid(gid)
+    os.setuid(uid)
+    os.environ.update(HOME="/home/agent", USER=AGENT_USER, LOGNAME=AGENT_USER)
+    logger.info("🔻 Dropped to unprivileged user %s (%s:%s)", AGENT_USER, uid, gid)
+
+
+if os.geteuid() == 0:
+    _drop_privileges(*_align_agent_with_host_user())
+
 shell = ShellCommunicator("bash")
 shell.start_session()
 

src/microbots/tools/internal_tool.py

@@ -57,7 +57,7 @@ def _setup_file_permission(env: Environment, file_copy: EnvFileCopies):
             # Convert to octal string for chmod (e.g., 7 -> "700" for owner rwx)
             # Set the assigned permission for owner, group and no permission for others
             permission_command = f"chmod {file_copy.permissions}{file_copy.permissions}0 {dest_path_in_container}"
-            output = env.execute(permission_command)
+            output = env.execute_privileged(permission_command)
             if output.return_code != 0:
                 logger.error(
                     "❌ Failed to set permission for file in container: %s to: %s",
@@ -87,7 +87,7 @@ def _copy_file_to_env(env: Environment, file_copy: EnvFileCopies):
                 # escape backslashes for shell execution
                 # content = content.replace('\\', '\\\\')
             dest_path_in_container = f"/{file_copy.dest}"
-            output = env.execute(
+            output = env.execute_privileged(
                 f'echo """{content}""" > {dest_path_in_container}'
             )
             if output.return_code != 0:
@@ -110,7 +110,7 @@ def _copy_file_to_env(env: Environment, file_copy: EnvFileCopies):
     def install_tool(self, env: Environment):
         logger.debug("Installing Internal tool: %s", self.name)
         for command in self.install_commands:
-            output = env.execute(command)
+            output = env.execute_privileged(command)
             if output.return_code != 0:
                 logger.error(
                     "❌ Failed to install tool: %s with command: %s\nOutput: %s",
@@ -164,7 +164,7 @@ def setup_tool(self, env: Environment):
     def uninstall_tool(self, env):
         super().uninstall_tool(env)
         for file_copy in self.files_to_copy:
-            output = env.execute(f"rm -f /{file_copy.dest}")
+            output = env.execute_privileged(f"rm -f /{file_copy.dest}")
             if output.return_code != 0:
                 logger.error(
                     "❌ Failed to remove copied file in container: %s during uninstallation of tool: %s",
@@ -176,7 +176,7 @@ def uninstall_tool(self, env):
                 )
 
         for command in self.uninstall_commands:
-            output = env.execute(command)
+            output = env.execute_privileged(command)
             if output.return_code != 0:
                 logger.error(
                     "❌ Failed to uninstall tool: %s with command: %s\nOutput: %s",

test/bot/test_copilot_bot.py

@@ -193,6 +193,7 @@ def mock_environment():
     success_return.stdout = "copilot version 1.0.0"
     success_return.stderr = ""
     env.execute = MagicMock(return_value=success_return)
+    env.execute_privileged = MagicMock(return_value=success_return)
     env.copy_to_container = MagicMock(return_value=True)
     env.stop = MagicMock()
     env.get_ipv4_address = MagicMock(return_value="172.17.0.2")
@@ -504,8 +505,12 @@ def test_install_cli_calls_execute(self, mock_environment):
                 github_token="ghp_test",
             )
             # _install_copilot_cli was called during __init__
-            # Verify that execute was called with npm install command
-            calls = [str(c) for c in mock_environment.execute.call_args_list]
+            # Verify that the install commands ran on the privileged channel
+            calls = [
+                str(c)
+                for c in mock_environment.execute.call_args_list
+                + mock_environment.execute_privileged.call_args_list
+            ]
             npm_calls = [c for c in calls if "npm install" in c or "copilot" in c]
             assert len(npm_calls) > 0, "Expected copilot-cli install commands"
             bot.stop()
@@ -518,6 +523,7 @@ def test_install_cli_raises_on_failure(self, mock_environment):
         fail_return.stdout = ""
         fail_return.stderr = "npm ERR! not found"
         mock_environment.execute = MagicMock(return_value=fail_return)
+        mock_environment.execute_privileged = MagicMock(return_value=fail_return)
 
         with (
             patch("microbots.bot.CopilotBot.get_free_port", side_effect=[9000]),
@@ -953,6 +959,7 @@ def side_effect(cmd, **kwargs):
             return success_ret
 
         mock_environment.execute = MagicMock(side_effect=side_effect)
+        mock_environment.execute_privileged = MagicMock(side_effect=side_effect)
 
         with (
             patch("microbots.bot.CopilotBot.get_free_port", side_effect=[9000]),