diff --git a/src/seclab_taskflows/mcp_servers/container_shell.py b/src/seclab_taskflows/mcp_servers/container_shell.py index 8039299..bed4fc7 100644 --- a/src/seclab_taskflows/mcp_servers/container_shell.py +++ b/src/seclab_taskflows/mcp_servers/container_shell.py @@ -1,6 +1,45 @@ # SPDX-FileCopyrightText: GitHub, Inc. # SPDX-License-Identifier: MIT +"""MCP server that runs shell commands inside a managed Docker container. + +Configuration is read from the process environment (set per toolbox in the +toolbox YAML's ``server_params.env`` block): + +- ``CONTAINER_IMAGE`` — image to run (required). +- ``CONTAINER_WORKSPACE`` — host path bind-mounted at ``/workspace`` (optional). +- ``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. +- ``CONTAINER_NETWORK`` — Docker network mode for the container. Defaults to + ``none`` so the container is egress-locked. Set it to ``bridge``, ``host``, or + a user-defined network to enable networking. + +Selecting a network mode from a toolbox: the agent passes only the toolbox's +declared ``env`` entries to this server, so a network mode is selectable at run +time only if the toolbox exposes the knob. To let callers opt in, add a +passthrough line to the toolbox ``env`` block, e.g.:: + + CONTAINER_NETWORK: "{{ env('CONTAINER_NETWORK', 'none') }}" + +A toolbox that needs networking by default (e.g. recon tooling) can use +``'bridge'`` as the template default instead. An empty or unset value always +falls back to ``none``, so isolation cannot be disabled by a blank variable. + +Transport: + +- ``CONTAINER_SHELL_TRANSPORT`` — MCP transport. Defaults to ``stdio`` for the + standard local case where the agent launches this server as a subprocess. Set + it to ``http``, ``streamable-http``, or ``sse`` to run as a network-accessible + MCP server that a remote agent can connect to. +- ``CONTAINER_SHELL_HOST`` / ``CONTAINER_SHELL_PORT`` — bind address for the + network transports (defaults ``127.0.0.1`` / ``8080``); ignored for ``stdio``. + +Which Docker daemon this server drives is orthogonal to the transport: the +docker CLI honours ``DOCKER_HOST`` from the environment, so pointing this server +at a specific (e.g. dedicated, isolated) daemon needs no code change here. +""" + import atexit import hashlib import json @@ -30,6 +69,25 @@ 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", "") +# Docker network mode for the container. Defaults to "none" so containers are +# egress-locked (no network access) unless a caller explicitly opts in by +# setting CONTAINER_NETWORK to a network name such as "bridge", "host", or a +# user-defined network. An empty or whitespace-only value falls back to "none" +# so the isolation default cannot be silently disabled by an unset variable. +CONTAINER_NETWORK = os.environ.get("CONTAINER_NETWORK", "none").strip() or "none" +# MCP transport selection. Defaults to "stdio" for the standard local case +# where the agent launches this server as a subprocess. Set +# CONTAINER_SHELL_TRANSPORT to "http", "streamable-http", or "sse" to run as a +# network-accessible server (for example, an isolated sidecar reached by a +# remote agent); CONTAINER_SHELL_HOST/PORT control the bind address for those +# transports and are ignored for stdio. +CONTAINER_SHELL_TRANSPORT = os.environ.get("CONTAINER_SHELL_TRANSPORT", "stdio").strip() or "stdio" +CONTAINER_SHELL_HOST = os.environ.get("CONTAINER_SHELL_HOST", "127.0.0.1").strip() or "127.0.0.1" +# Kept as a raw string and parsed to an int lazily in _run_server() so an +# invalid value only fails when a network transport is actually selected, not at +# import time under the default stdio transport (where host/port are ignored). +CONTAINER_SHELL_PORT = os.environ.get("CONTAINER_SHELL_PORT", "8080") +_SUPPORTED_TRANSPORTS = ("stdio", "http", "streamable-http", "sse") _DEFAULT_WORKDIR = "/workspace" _DOCKER_TIMEOUT = 30 @@ -38,11 +96,15 @@ def _persistent_name() -> str: """Derive a deterministic container name from the image for reuse across tasks. - Incorporates a hash of the full image reference (and optional - CONTAINER_PERSIST_KEY) to avoid collisions between long image names that - share a common prefix, or between independent runs of the same image. + Incorporates a hash of the full image reference, the configured network + mode, and an optional CONTAINER_PERSIST_KEY. 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. The hash also avoids collisions between long image names + that share a common prefix, or between independent runs of the same image. """ - key_material = CONTAINER_IMAGE + key_material = f"{CONTAINER_IMAGE}:net={CONTAINER_NETWORK}" if CONTAINER_PERSIST_KEY: key_material += f":{CONTAINER_PERSIST_KEY}" digest = hashlib.sha256(key_material.encode()).hexdigest()[:12] @@ -106,7 +168,7 @@ def _start_container() -> str: else: name = f"seclab-shell-{uuid.uuid4().hex[:8]}" - cmd = ["docker", "run", "-d", "--name", name] + cmd = ["docker", "run", "-d", "--name", name, "--network", CONTAINER_NETWORK] if not CONTAINER_PERSIST: cmd.append("--rm") if CONTAINER_WORKSPACE: @@ -149,7 +211,7 @@ def _stop_container() -> None: @mcp.tool() -def shell_exec( +def container_shell_exec( command: Annotated[str, Field(description="Shell command to execute inside the container")], timeout: Annotated[int, Field(description="Timeout in seconds")] = CONTAINER_TIMEOUT, workdir: Annotated[str, Field(description="Working directory inside the container")] = _DEFAULT_WORKDIR, @@ -176,5 +238,37 @@ def shell_exec( return output +def _run_server() -> None: + """Run the MCP server using the configured transport. + + Defaults to stdio (local subprocess use). When CONTAINER_SHELL_TRANSPORT + selects a network transport, the server binds CONTAINER_SHELL_HOST:PORT so a + remote agent can reach it. + """ + if CONTAINER_SHELL_TRANSPORT not in _SUPPORTED_TRANSPORTS: + msg = ( + f"Unsupported CONTAINER_SHELL_TRANSPORT {CONTAINER_SHELL_TRANSPORT!r}; " + f"expected one of {', '.join(_SUPPORTED_TRANSPORTS)}" + ) + raise ValueError(msg) + if CONTAINER_SHELL_TRANSPORT == "stdio": + mcp.run(show_banner=False) + else: + try: + port = int(CONTAINER_SHELL_PORT) + except (TypeError, ValueError) as exc: + msg = ( + f"Invalid CONTAINER_SHELL_PORT {CONTAINER_SHELL_PORT!r}; " + f"expected an integer for transport {CONTAINER_SHELL_TRANSPORT!r}" + ) + raise ValueError(msg) from exc + mcp.run( + transport=CONTAINER_SHELL_TRANSPORT, + host=CONTAINER_SHELL_HOST, + port=port, + show_banner=False, + ) + + if __name__ == "__main__": - mcp.run(show_banner=False) + _run_server() diff --git a/src/seclab_taskflows/taskflows/container_shell/README.md b/src/seclab_taskflows/taskflows/container_shell/README.md index 0257538..06e701f 100644 --- a/src/seclab_taskflows/taskflows/container_shell/README.md +++ b/src/seclab_taskflows/taskflows/container_shell/README.md @@ -1,7 +1,7 @@ # Container Shell Taskflows Runs arbitrary CLI commands inside an isolated Docker container. One container -per MCP server process — started on the first `shell_exec` call, stopped on +per MCP server process — started on the first `container_shell_exec` call, stopped on exit. An optional host directory is mounted at `/workspace` inside the container. Four container profiles are provided. Each has its own Dockerfile, toolbox @@ -117,17 +117,17 @@ taskflow: Analyse the binary at /workspace/target.elf using static analysis only. ``` -`shell_exec` requires user confirmation by default (`confirm: [shell_exec]` in +`container_shell_exec` requires user confirmation by default (`confirm: [container_shell_exec]` in all toolbox YAMLs). Pass `headless: true` at the task level to skip confirmation in automated pipelines. ## Notes -- The container is shared across all `shell_exec` calls within a single +- The container is shared across all `container_shell_exec` calls within a single taskflow run. State (files written, processes started) persists between calls. - `--rm` is set on `docker run`, so the container is removed automatically when stopped. - The container name follows the pattern `seclab-shell-<8 hex chars>` and is visible in `docker ps`. -- If `docker run` fails (e.g. image not found), `shell_exec` returns an error +- If `docker run` fails (e.g. image not found), `container_shell_exec` returns an error string rather than raising, so the agent can report the problem cleanly. diff --git a/src/seclab_taskflows/toolboxes/container_shell_base.yaml b/src/seclab_taskflows/toolboxes/container_shell_base.yaml index d5c1bf3..2883991 100644 --- a/src/seclab_taskflows/toolboxes/container_shell_base.yaml +++ b/src/seclab_taskflows/toolboxes/container_shell_base.yaml @@ -15,14 +15,15 @@ server_params: CONTAINER_TIMEOUT: "{{ env('CONTAINER_TIMEOUT', '30') }}" CONTAINER_PERSIST: "{{ env('CONTAINER_PERSIST', required=False) }}" CONTAINER_PERSIST_KEY: "{{ env('CONTAINER_PERSIST_KEY', required=False) }}" + CONTAINER_NETWORK: "{{ env('CONTAINER_NETWORK', required=False) }}" LOG_DIR: "{{ env('LOG_DIR') }}" confirm: - - shell_exec + - container_shell_exec server_prompt: | ## Container Shell (base) - You have access to an isolated Docker container. Use `shell_exec` to run commands. + You have access to an isolated Docker container. Use `container_shell_exec` to run commands. The working directory is /workspace (mapped from the host workspace if configured). Available tools in this container: diff --git a/src/seclab_taskflows/toolboxes/container_shell_malware_analysis.yaml b/src/seclab_taskflows/toolboxes/container_shell_malware_analysis.yaml index 0ed439b..f0f55cb 100644 --- a/src/seclab_taskflows/toolboxes/container_shell_malware_analysis.yaml +++ b/src/seclab_taskflows/toolboxes/container_shell_malware_analysis.yaml @@ -15,10 +15,11 @@ server_params: CONTAINER_TIMEOUT: "{{ env('CONTAINER_TIMEOUT', '60') }}" CONTAINER_PERSIST: "{{ env('CONTAINER_PERSIST', required=False) }}" CONTAINER_PERSIST_KEY: "{{ env('CONTAINER_PERSIST_KEY', required=False) }}" + CONTAINER_NETWORK: "{{ env('CONTAINER_NETWORK', required=False) }}" LOG_DIR: "{{ env('LOG_DIR') }}" confirm: - - shell_exec + - container_shell_exec server_prompt: | ## Container Shell (malware analysis) diff --git a/src/seclab_taskflows/toolboxes/container_shell_network_analysis.yaml b/src/seclab_taskflows/toolboxes/container_shell_network_analysis.yaml index 3d27a26..320f2f2 100644 --- a/src/seclab_taskflows/toolboxes/container_shell_network_analysis.yaml +++ b/src/seclab_taskflows/toolboxes/container_shell_network_analysis.yaml @@ -15,10 +15,11 @@ server_params: CONTAINER_TIMEOUT: "{{ env('CONTAINER_TIMEOUT', '30') }}" CONTAINER_PERSIST: "{{ env('CONTAINER_PERSIST', required=False) }}" CONTAINER_PERSIST_KEY: "{{ env('CONTAINER_PERSIST_KEY', required=False) }}" + CONTAINER_NETWORK: "{{ env('CONTAINER_NETWORK', 'bridge') }}" LOG_DIR: "{{ env('LOG_DIR') }}" confirm: - - shell_exec + - container_shell_exec server_prompt: | ## Container Shell (network analysis) diff --git a/src/seclab_taskflows/toolboxes/container_shell_remote.yaml b/src/seclab_taskflows/toolboxes/container_shell_remote.yaml new file mode 100644 index 0000000..8c8877e --- /dev/null +++ b/src/seclab_taskflows/toolboxes/container_shell_remote.yaml @@ -0,0 +1,44 @@ +# SPDX-FileCopyrightText: GitHub, Inc. +# SPDX-License-Identifier: MIT + +seclab-taskflow-agent: + filetype: toolbox + version: "1.0" + +# Remote variant of the container shell. Instead of launching container_shell +# as a local stdio subprocess (which needs a Docker socket in the agent +# environment), this connects to a container_shell MCP server that is already +# running elsewhere over Streamable HTTP. The server owns the Docker daemon; the +# agent only speaks MCP to it. This is what lets an AWF-isolated agent use +# containerized shell execution without ever being handed a Docker socket: the +# server runs outside the sandbox and is reached over a single allowlisted host +# service port. +# +# CONTAINER_SHELL_URL must point at the running server's Streamable HTTP +# endpoint, for example http://host.docker.internal:8765/mcp/. +server_params: + kind: streamable + url: "{{ env('CONTAINER_SHELL_URL') }}" + +confirm: + - container_shell_exec + +server_prompt: | + ## Container Shell (remote) + You have access to an isolated Docker container running on a remote shell + server. Use `container_shell_exec` to run commands inside it. The working + directory is /workspace (mapped from the host workspace if configured). + + Available tools in this container: + - bash, coreutils (ls, cat, grep, find, sed, awk, sort, uniq, wc, ...) + - file: identify file type by magic bytes + - strings: extract printable strings from binary files + - objdump: disassemble and dump object/binary files + - readelf: display ELF binary structure + - nm: list symbols from object files + - xxd / hexdump: hex inspection + - python3: scripting + - curl / wget: HTTP requests + - git: version control + + All commands run inside the container. Output includes stdout, stderr, and exit code. diff --git a/src/seclab_taskflows/toolboxes/container_shell_sast.yaml b/src/seclab_taskflows/toolboxes/container_shell_sast.yaml index d61a330..7fa6e48 100644 --- a/src/seclab_taskflows/toolboxes/container_shell_sast.yaml +++ b/src/seclab_taskflows/toolboxes/container_shell_sast.yaml @@ -15,10 +15,11 @@ server_params: CONTAINER_TIMEOUT: "{{ env('CONTAINER_TIMEOUT', '60') }}" CONTAINER_PERSIST: "{{ env('CONTAINER_PERSIST', required=False) }}" CONTAINER_PERSIST_KEY: "{{ env('CONTAINER_PERSIST_KEY', required=False) }}" + CONTAINER_NETWORK: "{{ env('CONTAINER_NETWORK', required=False) }}" LOG_DIR: "{{ env('LOG_DIR') }}" confirm: - - shell_exec + - container_shell_exec server_prompt: | ## Container Shell (SAST) diff --git a/tests/test_container_shell.py b/tests/test_container_shell.py index 96d29ed..7a8412b 100644 --- a/tests/test_container_shell.py +++ b/tests/test_container_shell.py @@ -1,6 +1,9 @@ # SPDX-FileCopyrightText: GitHub, Inc. # SPDX-License-Identifier: MIT +import atexit +import importlib +import os import subprocess from unittest.mock import MagicMock, patch @@ -28,6 +31,30 @@ def _reset_container(): cs_mod._container_name = None +def _reload_cs(): + """Reload cs_mod without accumulating atexit handlers. + + The module registers ``_stop_container`` with ``atexit`` at import time, so + reloading would stack a fresh handler each time. Unregister the current one + first so exactly one handler remains after the reload. + """ + atexit.unregister(cs_mod._stop_container) + return importlib.reload(cs_mod) + + +def _restore_env_and_reload(var, original): + """Restore an env var to its pre-test value and reload cs_mod. + + Reloading with the real environment (rather than the monkeypatched value) + keeps the module-level config from leaking into subsequent tests. + """ + if original is None: + os.environ.pop(var, None) + else: + os.environ[var] = original + _reload_cs() + + # --------------------------------------------------------------------------- # _start_container tests # --------------------------------------------------------------------------- @@ -89,16 +116,59 @@ def test_start_container_rejects_empty_image(self): with pytest.raises(RuntimeError, match="CONTAINER_IMAGE is not set"): cs_mod._start_container() + def test_start_container_default_network_none(self): + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_WORKSPACE", ""), + patch.object(cs_mod, "CONTAINER_NETWORK", "none"), + patch("subprocess.run", return_value=_make_proc(returncode=0)) as mock_run, + ): + cs_mod._start_container() + cmd = mock_run.call_args[0][0] + assert "--network" in cmd + assert cmd[cmd.index("--network") + 1] == "none" + + def test_start_container_opt_in_network(self): + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_WORKSPACE", ""), + patch.object(cs_mod, "CONTAINER_NETWORK", "bridge"), + patch("subprocess.run", return_value=_make_proc(returncode=0)) as mock_run, + ): + cs_mod._start_container() + cmd = mock_run.call_args[0][0] + assert "--network" in cmd + assert cmd[cmd.index("--network") + 1] == "bridge" + + def test_network_defaults_to_none_when_unset(self, monkeypatch): + original = os.environ.get("CONTAINER_NETWORK") + monkeypatch.delenv("CONTAINER_NETWORK", raising=False) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_NETWORK == "none" + finally: + _restore_env_and_reload("CONTAINER_NETWORK", original) + + @pytest.mark.parametrize("blank", ["", " ", "\t"]) + def test_network_falls_back_to_none_when_blank(self, monkeypatch, blank): + original = os.environ.get("CONTAINER_NETWORK") + monkeypatch.setenv("CONTAINER_NETWORK", blank) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_NETWORK == "none" + finally: + _restore_env_and_reload("CONTAINER_NETWORK", original) + # --------------------------------------------------------------------------- -# shell_exec tests +# container_shell_exec tests # --------------------------------------------------------------------------- class TestShellExec: def setup_method(self): _reset_container() - def test_shell_exec_lazy_start(self): + def test_container_shell_exec_lazy_start(self): start_proc = _make_proc(returncode=0) exec_proc = _make_proc(returncode=0, stdout="hello\n") with ( @@ -107,15 +177,15 @@ def test_shell_exec_lazy_start(self): patch("subprocess.run", side_effect=[start_proc, exec_proc]), ): assert cs_mod._container_name is None - result = cs_mod.shell_exec(command="echo hello") + result = cs_mod.container_shell_exec(command="echo hello") assert cs_mod._container_name is not None assert "hello" in result - def test_shell_exec_runs_command(self): + def test_container_shell_exec_runs_command(self): cs_mod._container_name = "seclab-shell-testtest" exec_proc = _make_proc(returncode=0, stdout="output\n") with patch("subprocess.run", return_value=exec_proc) as mock_run: - result = cs_mod.shell_exec(command="echo output", workdir="/workspace") + result = cs_mod.container_shell_exec(command="echo output", workdir="/workspace") cmd = mock_run.call_args[0][0] assert "docker" in cmd assert "exec" in cmd @@ -125,35 +195,35 @@ def test_shell_exec_runs_command(self): assert "echo output" in cmd assert "output" in result - def test_shell_exec_includes_exit_code(self): + def test_container_shell_exec_includes_exit_code(self): cs_mod._container_name = "seclab-shell-testtest" exec_proc = _make_proc(returncode=0, stdout="done\n") with patch("subprocess.run", return_value=exec_proc): - result = cs_mod.shell_exec(command="true") + result = cs_mod.container_shell_exec(command="true") assert "[exit code: 0]" in result - def test_shell_exec_nonzero_exit(self): + def test_container_shell_exec_nonzero_exit(self): cs_mod._container_name = "seclab-shell-testtest" exec_proc = _make_proc(returncode=1, stdout="", stderr="error\n") with patch("subprocess.run", return_value=exec_proc): - result = cs_mod.shell_exec(command="false") + result = cs_mod.container_shell_exec(command="false") assert "[exit code: 1]" in result assert "error" in result - def test_shell_exec_timeout(self): + def test_container_shell_exec_timeout(self): cs_mod._container_name = "seclab-shell-testtest" with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="docker", timeout=5)): - result = cs_mod.shell_exec(command="sleep 999", timeout=5) + result = cs_mod.container_shell_exec(command="sleep 999", timeout=5) assert "timeout" in result - def test_shell_exec_start_failure_returns_error(self): + def test_container_shell_exec_start_failure_returns_error(self): _reset_container() with ( patch.object(cs_mod, "CONTAINER_IMAGE", "bad-image:latest"), patch.object(cs_mod, "CONTAINER_WORKSPACE", ""), patch("subprocess.run", return_value=_make_proc(returncode=1, stderr="image not found")), ): - result = cs_mod.shell_exec(command="echo hi") + result = cs_mod.container_shell_exec(command="echo hi") assert "Failed to start container" in result assert cs_mod._container_name is None @@ -220,6 +290,17 @@ def test_persistent_name_differs_for_different_images(self): name_b = cs_mod._persistent_name() assert name_a != name_b + def test_persistent_name_varies_with_network(self): + with ( + patch.object(cs_mod, "CONTAINER_IMAGE", "test-image:latest"), + patch.object(cs_mod, "CONTAINER_PERSIST_KEY", ""), + ): + with patch.object(cs_mod, "CONTAINER_NETWORK", "none"): + name_none = cs_mod._persistent_name() + with patch.object(cs_mod, "CONTAINER_NETWORK", "bridge"): + name_bridge = cs_mod._persistent_name() + assert name_none != name_bridge + def test_start_reuses_running_persistent_container(self): inspect_proc = _make_proc( returncode=0, @@ -290,6 +371,98 @@ def test_is_running_returns_false_on_bad_json(self): assert cs_mod._is_running("test-name") is False +# --------------------------------------------------------------------------- +# Transport selection tests +# --------------------------------------------------------------------------- + +class TestRunServer: + def test_default_transport_is_stdio(self, monkeypatch): + original = os.environ.get("CONTAINER_SHELL_TRANSPORT") + monkeypatch.delenv("CONTAINER_SHELL_TRANSPORT", raising=False) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_SHELL_TRANSPORT == "stdio" + finally: + _restore_env_and_reload("CONTAINER_SHELL_TRANSPORT", original) + + def test_blank_transport_falls_back_to_stdio(self, monkeypatch): + original = os.environ.get("CONTAINER_SHELL_TRANSPORT") + monkeypatch.setenv("CONTAINER_SHELL_TRANSPORT", " ") + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_SHELL_TRANSPORT == "stdio" + finally: + _restore_env_and_reload("CONTAINER_SHELL_TRANSPORT", original) + + @pytest.mark.parametrize("blank", ["", " ", "\t"]) + def test_host_falls_back_to_default_when_blank(self, monkeypatch, blank): + original = os.environ.get("CONTAINER_SHELL_HOST") + monkeypatch.setenv("CONTAINER_SHELL_HOST", blank) + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_SHELL_HOST == "127.0.0.1" + finally: + _restore_env_and_reload("CONTAINER_SHELL_HOST", original) + + def test_run_server_stdio_uses_stdio_call(self): + with ( + patch.object(cs_mod, "CONTAINER_SHELL_TRANSPORT", "stdio"), + patch.object(cs_mod.mcp, "run") as mock_run, + ): + cs_mod._run_server() + mock_run.assert_called_once_with(show_banner=False) + + @pytest.mark.parametrize("transport", ["http", "streamable-http", "sse"]) + def test_run_server_network_transport_binds_host_port(self, transport): + with ( + patch.object(cs_mod, "CONTAINER_SHELL_TRANSPORT", transport), + patch.object(cs_mod, "CONTAINER_SHELL_HOST", "127.0.0.1"), + patch.object(cs_mod, "CONTAINER_SHELL_PORT", 9123), + patch.object(cs_mod.mcp, "run") as mock_run, + ): + cs_mod._run_server() + mock_run.assert_called_once_with( + transport=transport, + host="127.0.0.1", + port=9123, + show_banner=False, + ) + + def test_run_server_rejects_unknown_transport(self): + with ( + patch.object(cs_mod, "CONTAINER_SHELL_TRANSPORT", "carrier-pigeon"), + patch.object(cs_mod.mcp, "run") as mock_run, + ): + with pytest.raises(ValueError, match="Unsupported CONTAINER_SHELL_TRANSPORT"): + cs_mod._run_server() + mock_run.assert_not_called() + + def test_invalid_port_does_not_break_stdio_import(self, monkeypatch): + original = os.environ.get("CONTAINER_SHELL_PORT") + monkeypatch.setenv("CONTAINER_SHELL_PORT", "notaport") + try: + reloaded = _reload_cs() + assert reloaded.CONTAINER_SHELL_PORT == "notaport" + with ( + patch.object(reloaded, "CONTAINER_SHELL_TRANSPORT", "stdio"), + patch.object(reloaded.mcp, "run") as mock_run, + ): + reloaded._run_server() + mock_run.assert_called_once_with(show_banner=False) + finally: + _restore_env_and_reload("CONTAINER_SHELL_PORT", original) + + def test_invalid_port_raises_on_network_transport(self): + with ( + patch.object(cs_mod, "CONTAINER_SHELL_TRANSPORT", "http"), + patch.object(cs_mod, "CONTAINER_SHELL_PORT", "notaport"), + patch.object(cs_mod.mcp, "run") as mock_run, + ): + with pytest.raises(ValueError, match="Invalid CONTAINER_SHELL_PORT"): + cs_mod._run_server() + mock_run.assert_not_called() + + # --------------------------------------------------------------------------- # Toolbox YAML validation # ---------------------------------------------------------------------------