diff --git a/src/seclab_taskflows/mcp_servers/container_shell.py b/src/seclab_taskflows/mcp_servers/container_shell.py index 81f863e..d167fcd 100644 --- a/src/seclab_taskflows/mcp_servers/container_shell.py +++ b/src/seclab_taskflows/mcp_servers/container_shell.py @@ -8,6 +8,12 @@ - ``CONTAINER_IMAGE`` — image to run (required). - ``CONTAINER_WORKSPACE`` — host path bind-mounted at ``/workspace`` (optional). +- ``CONTAINER_WORKSPACE_MODE`` — bind-mount mode for that path, ``ro`` (default) + or ``rw``. Defaults to read-only so a command run inside the container cannot + modify the host's copy of the source under audit. Set it to ``rw`` only if a + taskflow genuinely needs to write into the workspace. An empty, unset, or + unrecognized value falls back to ``ro``, so the default cannot be silently + weakened by a blank variable. - ``CONTAINER_TIMEOUT`` — default per-command timeout in seconds (default 30). - ``CONTAINER_PERSIST`` — reuse a deterministic container across runs when truthy. - ``CONTAINER_PERSIST_KEY`` — extra key to distinguish persistent containers. @@ -66,6 +72,17 @@ CONTAINER_IMAGE = os.environ.get("CONTAINER_IMAGE", "") CONTAINER_WORKSPACE = os.environ.get("CONTAINER_WORKSPACE", "") +# Bind-mount mode for CONTAINER_WORKSPACE. Defaults to "ro" so a command run +# inside the container cannot modify the host's copy of the source under audit. +# This matters because the workspace is typically the tree being analyzed, and +# callers commonly collect it afterwards as an artifact: a writable mount makes +# that artifact agent-influenced rather than a faithful copy of the input. Set +# CONTAINER_WORKSPACE_MODE to "rw" to opt in to writes. An empty or +# unrecognized value falls back to "ro" so the default cannot be silently +# weakened by a blank variable. +CONTAINER_WORKSPACE_MODE = ( + "rw" if os.environ.get("CONTAINER_WORKSPACE_MODE", "").strip().lower() == "rw" else "ro" +) CONTAINER_TIMEOUT = int(os.environ.get("CONTAINER_TIMEOUT", "30")) CONTAINER_PERSIST = os.environ.get("CONTAINER_PERSIST", "").lower() in ("1", "true", "yes") CONTAINER_PERSIST_KEY = os.environ.get("CONTAINER_PERSIST_KEY", "") @@ -102,9 +119,14 @@ def _persistent_name() -> str: source trees. Including the network mode ensures a run configured for one network (e.g. the default "none") never reuses a persistent container that was created with a different, more permissive network (e.g. "bridge"), - which would otherwise silently re-enable egress. + which would otherwise silently re-enable egress. The workspace mount mode + is included for the same reason: a run configured for the default "ro" must + not reuse a container that was created with a writable workspace. """ - key_material = f"{CONTAINER_IMAGE}:{CONTAINER_WORKSPACE}:net={CONTAINER_NETWORK}" + key_material = ( + f"{CONTAINER_IMAGE}:{CONTAINER_WORKSPACE}" + f":net={CONTAINER_NETWORK}:ws={CONTAINER_WORKSPACE_MODE}" + ) if CONTAINER_PERSIST_KEY: key_material += f":{CONTAINER_PERSIST_KEY}" digest = hashlib.sha256(key_material.encode()).hexdigest()[:12] @@ -172,7 +194,7 @@ def _start_container() -> str: if not CONTAINER_PERSIST: cmd.append("--rm") if CONTAINER_WORKSPACE: - cmd += ["-v", f"{CONTAINER_WORKSPACE}:/workspace"] + cmd += ["-v", f"{CONTAINER_WORKSPACE}:/workspace:{CONTAINER_WORKSPACE_MODE}"] cmd += [CONTAINER_IMAGE, "tail", "-f", "/dev/null"] logging.debug(f"Starting container: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=_DOCKER_TIMEOUT) diff --git a/tests/test_container_shell.py b/tests/test_container_shell.py index ca1baad..17cc305 100644 --- a/tests/test_container_shell.py +++ b/tests/test_container_shell.py @@ -76,7 +76,7 @@ def test_start_container_success(self): assert "run" in cmd assert "--name" in cmd assert "-v" in cmd - assert "/host/workspace:/workspace" in cmd + assert "/host/workspace:/workspace:ro" in cmd assert "test-image:latest" in cmd assert "tail" in cmd @@ -91,6 +91,30 @@ def test_start_container_no_workspace(self): cmd = mock_run.call_args[0][0] assert "-v" not in cmd + def test_start_container_workspace_read_only_by_default(self): + """The workspace mount is read-only unless a caller opts in.""" + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_WORKSPACE", "/host/workspace"), + patch.object(cs_mod, "CONTAINER_WORKSPACE_MODE", "ro"), + patch("subprocess.run", return_value=_make_proc(returncode=0)) as mock_run, + ): + cs_mod._start_container() + cmd = mock_run.call_args[0][0] + assert "/host/workspace:/workspace:ro" in cmd + + def test_start_container_workspace_rw_opt_in(self): + """CONTAINER_WORKSPACE_MODE=rw restores a writable mount.""" + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_WORKSPACE", "/host/workspace"), + patch.object(cs_mod, "CONTAINER_WORKSPACE_MODE", "rw"), + patch("subprocess.run", return_value=_make_proc(returncode=0)) as mock_run, + ): + cs_mod._start_container() + cmd = mock_run.call_args[0][0] + assert "/host/workspace:/workspace:rw" in cmd + def test_start_container_failure(self): with ( patch.object(cs_mod, "CONTAINER_IMAGE", "missing-image:latest"), @@ -140,6 +164,36 @@ def test_start_container_opt_in_network(self): assert "--network" in cmd assert cmd[cmd.index("--network") + 1] == "bridge" + def test_workspace_mode_defaults_to_ro_when_unset(self, monkeypatch): + original = os.environ.get("CONTAINER_WORKSPACE_MODE") + monkeypatch.delenv("CONTAINER_WORKSPACE_MODE", raising=False) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_WORKSPACE_MODE == "ro" + finally: + _restore_env_and_reload("CONTAINER_WORKSPACE_MODE", original) + + @pytest.mark.parametrize("value", ["", " ", "\t", "bogus", "readwrite", "ro"]) + def test_workspace_mode_falls_back_to_ro(self, monkeypatch, value): + """Only an explicit "rw" opts in; anything else is read-only.""" + original = os.environ.get("CONTAINER_WORKSPACE_MODE") + monkeypatch.setenv("CONTAINER_WORKSPACE_MODE", value) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_WORKSPACE_MODE == "ro" + finally: + _restore_env_and_reload("CONTAINER_WORKSPACE_MODE", original) + + @pytest.mark.parametrize("value", ["rw", "RW", " rw "]) + def test_workspace_mode_rw_opt_in(self, monkeypatch, value): + original = os.environ.get("CONTAINER_WORKSPACE_MODE") + monkeypatch.setenv("CONTAINER_WORKSPACE_MODE", value) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_WORKSPACE_MODE == "rw" + finally: + _restore_env_and_reload("CONTAINER_WORKSPACE_MODE", original) + def test_network_defaults_to_none_when_unset(self, monkeypatch): original = os.environ.get("CONTAINER_NETWORK") monkeypatch.delenv("CONTAINER_NETWORK", raising=False) @@ -301,6 +355,19 @@ def test_persistent_name_differs_for_different_workspaces(self): name_b = cs_mod._persistent_name() assert name_a != name_b + def test_persistent_name_varies_with_workspace_mode(self): + """A "ro" run must not reuse a container created with a writable mount.""" + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_WORKSPACE", "/source/tree"), + patch.object(cs_mod, "CONTAINER_PERSIST_KEY", ""), + ): + with patch.object(cs_mod, "CONTAINER_WORKSPACE_MODE", "ro"): + name_ro = cs_mod._persistent_name() + with patch.object(cs_mod, "CONTAINER_WORKSPACE_MODE", "rw"): + name_rw = cs_mod._persistent_name() + assert name_ro != name_rw + def test_persistent_name_varies_with_network(self): with ( patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"),