Mount CONTAINER_WORKSPACE read-only by default

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)

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"),