From e595de8dddd2658e15b2cc79fb5962a1007f7177 Mon Sep 17 00:00:00 2001 From: bala Date: Wed, 23 Sep 2026 07:36:17 +0000 Subject: [PATCH 1/2] Remove sudo access to agent shell --- src/microbots/bot/CopilotBot.py | 6 +- src/microbots/environment/Environment.py | 6 + .../local_docker/LocalDockerEnvironment.py | 124 +++++++++++++----- .../local_docker/image_builder/Dockerfile | 12 +- .../image_builder/ShellCommunicator.py | 2 +- .../local_docker/image_builder/dockerShell.py | 40 ++++++ src/microbots/tools/internal_tool.py | 10 +- test/bot/test_copilot_bot.py | 11 +- test/bot/test_microbot.py | 6 +- test/tools/test_tool.py | 9 ++ 10 files changed, 182 insertions(+), 44 deletions(-) diff --git a/src/microbots/bot/CopilotBot.py b/src/microbots/bot/CopilotBot.py index bcb3a13d..88d91b23 100644 --- a/src/microbots/bot/CopilotBot.py +++ b/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}" diff --git a/src/microbots/environment/Environment.py b/src/microbots/environment/Environment.py index 83df8716..68e34a34 100644 --- a/src/microbots/environment/Environment.py +++ b/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. " diff --git a/src/microbots/environment/local_docker/LocalDockerEnvironment.py b/src/microbots/environment/local_docker/LocalDockerEnvironment.py index d91e98ea..255dd4c7 100644 --- a/src/microbots/environment/local_docker/LocalDockerEnvironment.py +++ b/src/microbots/environment/local_docker/LocalDockerEnvironment.py @@ -79,8 +79,9 @@ 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, @@ -88,8 +89,15 @@ def start(self): 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("<", "<").replace(">", ">") 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", "" 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 diff --git a/src/microbots/environment/local_docker/image_builder/Dockerfile b/src/microbots/environment/local_docker/image_builder/Dockerfile index 4356884f..8ae9e1ed 100644 --- a/src/microbots/environment/local_docker/image_builder/Dockerfile +++ b/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 . diff --git a/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py b/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py index 4a8b71dc..0aed4d9f 100644 --- a/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py +++ b/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: diff --git a/src/microbots/environment/local_docker/image_builder/dockerShell.py b/src/microbots/environment/local_docker/image_builder/dockerShell.py index 1575c594..97335b1a 100644 --- a/src/microbots/environment/local_docker/image_builder/dockerShell.py +++ b/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() diff --git a/src/microbots/tools/internal_tool.py b/src/microbots/tools/internal_tool.py index 7292ff96..76a3a6f5 100644 --- a/src/microbots/tools/internal_tool.py +++ b/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", diff --git a/test/bot/test_copilot_bot.py b/test/bot/test_copilot_bot.py index d175ccbc..9d7d2b62 100644 --- a/test/bot/test_copilot_bot.py +++ b/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]), diff --git a/test/bot/test_microbot.py b/test/bot/test_microbot.py index 151fabda..e414f025 100644 --- a/test/bot/test_microbot.py +++ b/test/bot/test_microbot.py @@ -8,6 +8,7 @@ import os from pathlib import Path from pprint import pformat +import shutil import subprocess import sys from unittest.mock import patch, Mock, MagicMock @@ -64,7 +65,7 @@ def log_file_path(self, tmpdir: Path): assert tmpdir.exists() yield tmpdir / "error.log" if tmpdir.exists(): - subprocess.run(["sudo", "rm", "-rf", str(tmpdir)]) + shutil.rmtree(str(tmpdir), ignore_errors=True) @pytest.fixture(scope="function") def ro_mount(self, test_repo: Path): @@ -500,6 +501,7 @@ def test_tool_usage_instructions_appended_to_system_prompt(self): # Create a mock environment mock_env = Mock() mock_env.execute.return_value = Mock(return_code=0, stdout="", stderr="") + mock_env.execute_privileged = mock_env.execute # Mock the environment and LLM creation to avoid actual Docker/API calls with patch('microbots.llm.azure_openai_api.api_key', 'test-key'), \ @@ -549,6 +551,7 @@ def test_multiple_tool_usage_instructions_appended(self): # Create a mock environment mock_env = Mock() mock_env.execute.return_value = Mock(return_code=0, stdout="", stderr="") + mock_env.execute_privileged = mock_env.execute # Mock the environment and LLM creation with patch('microbots.llm.anthropic_api.Anthropic'): @@ -1044,6 +1047,7 @@ def test_tool_usage_instructions_in_system_prompt(self, cscope_tool): """Test that tool usage instructions are appended to the bot's system prompt.""" mock_env = Mock() mock_env.execute.return_value = Mock(return_code=0, stdout="cscope: version 15.9", stderr="") + mock_env.execute_privileged = mock_env.execute base_prompt = "You are a code analysis assistant." diff --git a/test/tools/test_tool.py b/test/tools/test_tool.py index eb4672b5..8f5f5d4f 100644 --- a/test/tools/test_tool.py +++ b/test/tools/test_tool.py @@ -19,6 +19,15 @@ from microbots.tools.tool import EnvFileCopies +@pytest.fixture(autouse=True) +def alias_privileged_channel(monkeypatch): + """Point the root control channel at ``execute`` on every Mock, so the + command-sequence assertions below cover both channels.""" + monkeypatch.setattr( + Mock, "execute_privileged", property(lambda self: self.execute), raising=False + ) + + @pytest.mark.unit class TestToolOptionalArguments: """Unit tests for Tool class optional arguments handling.""" From c6fd492590c09fd77a2036c40d13893bb427fecf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 23 Sep 2026 08:50:53 +0000 Subject: [PATCH 2/2] test: cover all privilege-removal patch lines Co-authored-by: 0xba1a <2942888+0xba1a@users.noreply.github.com> --- .../local_docker/test_docker_shell.py | 143 +++++++++++++++ .../test_local_docker_environment.py | 168 +++++++++++++++++- test/environment/test_environment.py | 30 ++++ 3 files changed, 340 insertions(+), 1 deletion(-) create mode 100644 test/environment/local_docker/test_docker_shell.py create mode 100644 test/environment/test_environment.py diff --git a/test/environment/local_docker/test_docker_shell.py b/test/environment/local_docker/test_docker_shell.py new file mode 100644 index 00000000..0b2908e4 --- /dev/null +++ b/test/environment/local_docker/test_docker_shell.py @@ -0,0 +1,143 @@ +"""Unit tests for privilege removal before the container shell starts.""" + +import importlib.util +import logging +import os +from pathlib import Path +import runpy +import subprocess +import sys +from unittest.mock import Mock, call, patch + +import pytest + +from microbots.environment.local_docker import image_builder + + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def shell_runtime(monkeypatch): + builder = Path(image_builder.__file__).parent + runtime = Mock() + runtime.geteuid.return_value = 0 + for name in ("geteuid", "setgroups", "setgid", "setuid"): + monkeypatch.setattr(os, name, getattr(runtime, name)) + monkeypatch.setattr(subprocess, "run", runtime.run) + + with patch.dict(os.environ), patch("logging.basicConfig") as configure_logging: + for name in ("AGENT_UID", "AGENT_GID", "BOT_WORKDIR"): + monkeypatch.delenv(name, raising=False) + os.environ.update(HOME="/root", USER="root", LOGNAME="root") + + # Load the real communicator to cover its logging configuration, but + # replace shell creation so no subprocesses or threads are started. + spec = importlib.util.spec_from_file_location( + "ShellCommunicator", builder / "ShellCommunicator.py" + ) + communicator = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "ShellCommunicator", communicator) + spec.loader.exec_module(communicator) + monkeypatch.setattr(communicator, "ShellCommunicator", runtime.shell) + + yield lambda: runpy.run_path(str(builder / "dockerShell.py")), runtime, configure_logging + + +def test_communicator_logs_to_agent_owned_directory(shell_runtime): + _, _, configure_logging = shell_runtime + + configure_logging.assert_called_once_with( + level=logging.DEBUG, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + filename="/var/log/microbots/ShellCommunicator.log", + ) + + +@pytest.mark.parametrize( + "uid, gid, workdir, expected_uid, expected_gid", + [ + (None, None, None, 1001, 1001), + ("", "", "", 1001, 1001), + ("1001", "1001", None, 1001, 1001), + ("2001", "2002", "/work dir", 2001, 2002), + ("1001", "2002", None, 1001, 2002), + ], +) +def test_root_startup_drops_privileges_before_starting_shell( + shell_runtime, monkeypatch, uid, gid, workdir, expected_uid, expected_gid +): + load_shell, runtime, _ = shell_runtime + for name, value in (("AGENT_UID", uid), ("AGENT_GID", gid), ("BOT_WORKDIR", workdir)): + if value is not None: + monkeypatch.setenv(name, value) + + load_shell() + + expected_calls = [call.geteuid()] + if (expected_uid, expected_gid) != (1001, 1001): + expected_calls.extend([ + call.run(["groupmod", "-g", str(expected_gid), "agent"], check=True), + call.run( + ["usermod", "-u", str(expected_uid), "-g", str(expected_gid), "agent"], + check=True, + ), + ]) + paths = ["/home/agent", "/var/log/microbots"] + if workdir: + paths.append(workdir) + expected_calls.extend( + call.run(["chown", "-R", f"{expected_uid}:{expected_gid}", path], check=False) + for path in paths + ) + expected_calls.extend([ + call.setgroups([expected_gid]), + call.setgid(expected_gid), + call.setuid(expected_uid), + call.shell("bash"), + call.shell().start_session(), + ]) + assert runtime.mock_calls == expected_calls + assert os.environ["HOME"] == "/home/agent" + assert os.environ["USER"] == os.environ["LOGNAME"] == "agent" + + +def test_non_root_startup_preserves_identity(shell_runtime): + load_shell, runtime, _ = shell_runtime + runtime.geteuid.return_value = 1001 + os.environ.update(HOME="/home/existing", USER="existing", LOGNAME="existing") + + load_shell() + + assert runtime.mock_calls == [ + call.geteuid(), + call.shell("bash"), + call.shell().start_session(), + ] + assert os.environ["HOME"] == "/home/existing" + assert os.environ["USER"] == os.environ["LOGNAME"] == "existing" + + +@pytest.mark.parametrize("operation", ["setgroups", "setgid", "setuid"]) +def test_privilege_drop_failure_prevents_shell_start(shell_runtime, operation): + load_shell, runtime, _ = shell_runtime + getattr(runtime, operation).side_effect = PermissionError("cannot drop privileges") + + with pytest.raises(PermissionError, match="cannot drop privileges"): + load_shell() + + runtime.shell.assert_not_called() + + +def test_identity_alignment_failure_prevents_shell_start(shell_runtime, monkeypatch): + load_shell, runtime, _ = shell_runtime + monkeypatch.setenv("AGENT_UID", "2001") + runtime.run.side_effect = subprocess.CalledProcessError(1, "groupmod") + + with pytest.raises(subprocess.CalledProcessError): + load_shell() + + runtime.setgroups.assert_not_called() + runtime.setgid.assert_not_called() + runtime.setuid.assert_not_called() + runtime.shell.assert_not_called() diff --git a/test/environment/local_docker/test_local_docker_environment.py b/test/environment/local_docker/test_local_docker_environment.py index e67f4956..71d31125 100644 --- a/test/environment/local_docker/test_local_docker_environment.py +++ b/test/environment/local_docker/test_local_docker_environment.py @@ -7,7 +7,7 @@ import re import logging import time -from unittest.mock import patch, Mock, MagicMock +from unittest.mock import patch, Mock, MagicMock, call # Add src to path for imports import sys @@ -15,6 +15,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src"))) from microbots.environment.local_docker.LocalDockerEnvironment import LocalDockerEnvironment +from microbots.environment.Environment import CmdReturn from microbots.extras.mount import Mount from microbots.constants import DOCKER_WORKING_DIR, WORKING_DIR @@ -543,3 +544,168 @@ def test_raises_runtime_error_when_ip_is_empty(self): with pytest.raises(RuntimeError, match="Could not determine container IP address"): env.get_ipv4_address() + + +@pytest.mark.unit +class TestPrivilegedControlChannel: + """Unit tests for privileged setup isolated from the bot's shell.""" + + @pytest.fixture + def env(self): + env = LocalDockerEnvironment.__new__(LocalDockerEnvironment) + env.deleted = True + env.container = Mock() + env.execute = Mock() + env.overlay_mount = False + env.folder_to_mount = Mock(sandbox_path=f"{DOCKER_WORKING_DIR}/repo") + return env + + @pytest.mark.parametrize( + "uid, expected", + [ + (None, {}), + (0, {}), + (2001, {"AGENT_UID": "2001", "AGENT_GID": "2002"}), + ], + ) + def test_host_identity(self, monkeypatch, uid, expected): + if uid is None: + monkeypatch.delattr(os, "getuid") + else: + monkeypatch.setattr(os, "getuid", lambda: uid) + monkeypatch.setattr(os, "getgid", Mock(return_value=2002)) + + assert LocalDockerEnvironment._host_identity() == expected + if not expected: + os.getgid.assert_not_called() + + def test_start_restricts_container_privileges_and_published_port(self, env): + env.image = "test_image" + env.working_dir = "/tmp/test_workdir" + env.folder_to_mount = None + env.port = 12345 + env.container_port = 8080 + env.client = Mock() + env.client.containers.run.return_value.id = "test_container" + identity = {"AGENT_UID": "2001", "AGENT_GID": "2002"} + + with patch.object(env, "_host_identity", return_value=identity), \ + patch("microbots.environment.local_docker.LocalDockerEnvironment.time.sleep"): + env.start() + + env.client.containers.run.assert_called_once_with( + "test_image", + volumes={"/tmp/test_workdir": {"bind": DOCKER_WORKING_DIR, "mode": "rw"}}, + ports={"8080/tcp": ("127.0.0.1", 12345)}, + detach=True, + working_dir="/app", + cap_add=["SYS_ADMIN"], + security_opt=["no-new-privileges:true", "apparmor=unconfined"], + environment={"BOT_PORT": "8080", "BOT_WORKDIR": DOCKER_WORKING_DIR, **identity}, + ) + assert env.container is env.client.containers.run.return_value + env.execute.assert_called_once_with("cd /") + + @pytest.mark.parametrize( + "exit_code, stdout, stderr, expected", + [ + (0, b"output", b"", CmdReturn("output", "", 0)), + (7, b"out\xff", b"err\xff", CmdReturn("out\ufffd", "err\ufffd", 7)), + (None, None, None, CmdReturn("", "", 0)), + ], + ) + def test_execute_privileged_uses_root_docker_exec(self, env, exit_code, stdout, stderr, expected): + env.container.exec_run.return_value = (exit_code, (stdout, stderr)) + + result = env.execute_privileged("echo test", timeout=15) + + env.container.exec_run.assert_called_once_with( + ["bash", "-lc", "echo test"], user="root", demux=True + ) + env.execute.assert_not_called() + assert result == expected + + def test_execute_privileged_requires_container(self, env): + env.container = None + + with pytest.raises(RuntimeError, match="No active container"): + env.execute_privileged("echo test") + + env.execute.assert_not_called() + + @pytest.mark.parametrize("sensitive", [False, True]) + def test_execute_privileged_logging(self, env, caplog, sensitive): + env.container.exec_run.return_value = (0, (None, None)) + + with caplog.at_level(logging.DEBUG): + env.execute_privileged("echo private-value", sensitive=sensitive) + + assert ("echo private-value" in caplog.text) is not sensitive + assert ("" in caplog.text) is sensitive + env.container.exec_run.assert_called_once_with( + ["bash", "-lc", "echo private-value"], user="root", demux=True + ) + + @pytest.mark.parametrize("identity", [{}, {"AGENT_UID": "2001", "AGENT_GID": "2002"}]) + def test_setup_overlay_uses_privileged_channel(self, env, identity): + env.execute_privileged = Mock(return_value=CmdReturn("", "", 0)) + sandbox = env.folder_to_mount.sandbox_path + overlay = f"{DOCKER_WORKING_DIR}/overlay/repo" + + with patch.object(env, "_host_identity", return_value=identity): + env._setup_overlay_mount() + + expected_calls = [ + call( + f"mkdir -p {sandbox} {overlay}/upper {overlay}/work && " + f"mount -t overlay overlay " + f"-o lowerdir=/ro/repo/,upperdir={overlay}/upper/,workdir={overlay}/work/ " + f"{sandbox}" + ) + ] + if identity: + expected_calls.append(call(f"chown 2001:2002 {sandbox}")) + assert env.execute_privileged.call_args_list == expected_calls + assert env.overlay_mount is True + env.execute.assert_not_called() + + def test_setup_overlay_failure_does_not_mark_mounted(self, env): + env.execute_privileged = Mock(return_value=CmdReturn("", "mount denied", 1)) + + with pytest.raises(RuntimeError, match="Failed to set up overlay mount for repo: mount denied"): + env._setup_overlay_mount() + + env.execute_privileged.assert_called_once() + assert env.overlay_mount is False + env.execute.assert_not_called() + + @pytest.mark.parametrize("unmount_code, remove_code", [(0, 0), (1, 0), (0, 1)]) + def test_teardown_overlay_cleans_up_and_resets_state(self, env, caplog, unmount_code, remove_code): + env.overlay_mount = True + env.execute_privileged = Mock(side_effect=[ + CmdReturn("", "unmount denied" if unmount_code else "", unmount_code), + CmdReturn("", "remove denied" if remove_code else "", remove_code), + ]) + + with caplog.at_level(logging.INFO): + env._teardown_overlay_mount() + + assert env.execute_privileged.call_args_list == [ + call(f"umount -l {DOCKER_WORKING_DIR}/repo"), + call(f"rm -rf {DOCKER_WORKING_DIR}/repo {DOCKER_WORKING_DIR}/overlay/repo"), + ] + assert ("Failed to unmount overlay: unmount denied" in caplog.text) == bool(unmount_code) + assert ("Failed to remove overlay directories: remove denied" in caplog.text) == bool(remove_code) + assert env.overlay_mount is False + env.execute.assert_not_called() + + def test_teardown_overlay_exception_resets_state(self, env, caplog): + env.overlay_mount = True + env.execute_privileged = Mock(side_effect=RuntimeError("container unavailable")) + + with caplog.at_level(logging.ERROR): + env._teardown_overlay_mount() + + assert "Failed to teardown overlay mount: container unavailable" in caplog.text + assert env.overlay_mount is False + env.execute.assert_not_called() diff --git a/test/environment/test_environment.py b/test/environment/test_environment.py new file mode 100644 index 00000000..29e0e801 --- /dev/null +++ b/test/environment/test_environment.py @@ -0,0 +1,30 @@ +"""Unit tests for the default environment control channel.""" + +from unittest.mock import Mock + +import pytest + +from microbots.environment.Environment import CmdReturn, Environment + + +@pytest.mark.unit +@pytest.mark.parametrize( + "options, timeout, sensitive", + [ + ({}, 300, False), + ({"timeout": 15, "sensitive": True}, 15, True), + ({"timeout": None}, None, False), + ], +) +def test_execute_privileged_delegates_to_execute(options, timeout, sensitive): + class CustomEnvironment(Environment): + start = Mock() + stop = Mock() + execute = Mock(return_value=CmdReturn("output", "error", 7)) + + env = CustomEnvironment() + + result = env.execute_privileged("echo test", **options) + + env.execute.assert_called_once_with("echo test", timeout=timeout, sensitive=sensitive) + assert result is env.execute.return_value