Use shell=True for Windows CMD URL escaping

scripts/setup_environment.py

@@ -3260,29 +3260,6 @@ def verify_nodejs_available() -> bool:
     return False
 
 
-def escape_for_cmd(arg: str) -> str:
-    """Escape special characters for Windows CMD execution.
-
-    When subprocess.run() is called with a .CMD file on Windows, CMD interprets
-    special characters like & as command separators. This function escapes such
-    characters with the ^ prefix to ensure they are treated as literal characters.
-
-    Args:
-        arg: The argument string to escape.
-
-    Returns:
-        The escaped string safe for CMD execution.
-    """
-    # CMD special characters that need escaping with ^
-    # IMPORTANT: ^ must be escaped FIRST to avoid double-escaping the ^ characters
-    # that are inserted when escaping other special characters
-    special_chars = ['^', '&', '|', '<', '>', '%']
-    result = arg
-    for char in special_chars:
-        result = result.replace(char, f'^{char}')
-    return result
-
-
 def configure_mcp_server(server: dict[str, Any]) -> bool:
     """Configure a single MCP server."""
     name = server.get('name')
@@ -3402,15 +3379,18 @@ def configure_mcp_server(server: dict[str, Any]) -> bool:
                     capture_output=True,
                 )
 
-                # Also try with direct execution
+                # Also try with direct execution using shell=True
                 if result.returncode != 0:
                     info('Trying direct execution...')
-                    # Escape URL for CMD execution to handle special characters like &
-                    cmd_escaped = base_cmd.copy()
-                    if url in cmd_escaped:
-                        url_index = cmd_escaped.index(url)
-                        cmd_escaped[url_index] = escape_for_cmd(url)
-                    result = run_command(cmd_escaped, capture_output=True)
+                    # Use shell=True with double-quoted URL for Windows CMD compatibility
+                    # This ensures & in URLs is not interpreted as a command separator
+                    url_quoted = f'"{url}"'
+                    header_part = f' --header "{header}"' if header else ''
+                    cmd_str = (
+                        f'"{claude_cmd}" mcp add --scope {scope} {name}{env_part} '
+                        f'--transport {transport}{header_part} {url_quoted}'
+                    )
+                    result = subprocess.run(cmd_str, shell=True, capture_output=True, text=True)
             else:
                 # On Unix, spawn new bash with updated PATH
                 parent_dir = Path(claude_cmd).parent
@@ -3474,13 +3454,20 @@ def configure_mcp_server(server: dict[str, Any]) -> bool:
         time.sleep(2)
 
         # Direct execution with full path
-        # On Windows, escape URL for CMD execution to handle special characters like &
-        if sys.platform == 'win32' and url and url in base_cmd:
-            cmd_escaped = base_cmd.copy()
-            url_index = cmd_escaped.index(url)
-            cmd_escaped[url_index] = escape_for_cmd(url)
-            info(f"Retrying with direct command: {' '.join(str(arg) for arg in cmd_escaped)}")
-            result = run_command(cmd_escaped, capture_output=False)  # Show output for debugging
+        # On Windows with HTTP transport, use shell=True with double-quoted URL
+        if sys.platform == 'win32' and transport and url:
+            # Use shell=True with double-quoted URL for Windows CMD compatibility
+            # This ensures & in URLs is not interpreted as a command separator
+            url_quoted = f'"{url}"'
+            env_flags = ' '.join(f'--env "{e}"' for e in env_list) if env_list else ''
+            env_part_retry = f' {env_flags}' if env_flags else ''
+            header_part = f' --header "{header}"' if header else ''
+            cmd_str = (
+                f'"{claude_cmd}" mcp add --scope {scope} {name}{env_part_retry} '
+                f'--transport {transport}{header_part} {url_quoted}'
+            )
+            info(f'Retrying with direct command: {cmd_str}')
+            result = subprocess.run(cmd_str, shell=True, capture_output=False, text=True)
         else:
             info(f"Retrying with direct command: {' '.join(str(arg) for arg in base_cmd)}")
             result = run_command(base_cmd, capture_output=False)  # Show output for debugging

tests/test_setup_environment.py

@@ -164,52 +164,6 @@ def test_find_command(self, mock_which):
         assert setup_environment.find_command('git') == '/usr/bin/git'
 
 
-class TestEscapeForCmd:
-    """Test CMD special character escaping function."""
-
-    def test_escape_ampersand(self):
-        """Test escaping ampersand character."""
-        url = 'https://example.com?param1=value1&param2=value2'
-        escaped = setup_environment.escape_for_cmd(url)
-        assert escaped == 'https://example.com?param1=value1^&param2=value2'
-
-    def test_escape_multiple_special_chars(self):
-        """Test escaping multiple special characters."""
-        arg = 'test&value|with<special>chars^and%percent'
-        escaped = setup_environment.escape_for_cmd(arg)
-        assert escaped == 'test^&value^|with^<special^>chars^^and^%percent'
-
-    def test_escape_empty_string(self):
-        """Test escaping empty string returns empty string."""
-        assert setup_environment.escape_for_cmd('') == ''
-
-    def test_escape_no_special_chars(self):
-        """Test escaping string without special chars returns unchanged."""
-        url = 'https://example.com/path/to/resource'
-        assert setup_environment.escape_for_cmd(url) == url
-
-    def test_escape_supabase_url(self):
-        """Test escaping real-world Supabase MCP URL with & query parameter."""
-        url = 'https://mcp.supabase.com/mcp?project_ref=xkeujjyicwuxwxelliae&read_only=true'
-        escaped = setup_environment.escape_for_cmd(url)
-        assert escaped == 'https://mcp.supabase.com/mcp?project_ref=xkeujjyicwuxwxelliae^&read_only=true'
-
-    def test_escape_url_with_multiple_query_params(self):
-        """Test escaping URL with multiple ampersands in query string."""
-        url = 'https://api.example.com?key=abc&token=xyz&mode=read&format=json'
-        escaped = setup_environment.escape_for_cmd(url)
-        expected = 'https://api.example.com?key=abc^&token=xyz^&mode=read^&format=json'
-        assert escaped == expected
-
-    def test_escape_already_escaped(self):
-        """Test that already escaped characters get double-escaped."""
-        # This is correct behavior - if someone passes already escaped string,
-        # we escape the caret too
-        arg = 'test^&value'
-        escaped = setup_environment.escape_for_cmd(arg)
-        assert escaped == 'test^^^&value'
-
-
 class TestTildeExpansion:
     """Test tilde expansion in commands."""
 
@@ -1155,6 +1109,105 @@ def test_configure_mcp_server_http_with_header_and_special_url(self, mock_run, m
         assert 'client_id=123' in add_cmd_str
         assert 'scope=read' in add_cmd_str
 
+    @patch('platform.system')
+    @patch('subprocess.run')
+    @patch('setup_environment.run_command')
+    @patch('setup_environment.find_command_robust')
+    def test_configure_mcp_server_windows_fallback_uses_shell_true(
+        self, mock_find, mock_run_cmd, mock_subprocess_run, mock_system,
+    ):
+        """Test Windows fallback uses shell=True with double-quoted URL.
+
+        When PowerShell path fails on Windows, the fallback should use subprocess.run
+        with shell=True and the URL wrapped in double quotes to prevent & from being
+        interpreted as a command separator.
+        """
+        mock_system.return_value = 'Windows'
+        mock_find.return_value = 'C:\\Users\\Test\\AppData\\Roaming\\npm\\claude.CMD'
+
+        # First 3 calls are for removing from scopes (success)
+        # 4th call is PowerShell which fails
+        mock_run_cmd.side_effect = [
+            subprocess.CompletedProcess([], 0, '', ''),  # remove user
+            subprocess.CompletedProcess([], 0, '', ''),  # remove local
+            subprocess.CompletedProcess([], 0, '', ''),  # remove project
+            subprocess.CompletedProcess([], 1, '', 'PowerShell failed'),  # PowerShell fails
+        ]
+
+        # subprocess.run (shell=True) succeeds
+        mock_subprocess_run.return_value = subprocess.CompletedProcess([], 0, '', '')
+
+        server = {
+            'name': 'supabase',
+            'scope': 'user',
+            'transport': 'http',
+            'url': 'https://mcp.supabase.com/mcp?project_ref=xxx&read_only=true',
+        }
+
+        result = setup_environment.configure_mcp_server(server)
+        assert result is True
+
+        # Verify subprocess.run was called with shell=True
+        assert mock_subprocess_run.called
+        call_args = mock_subprocess_run.call_args
+
+        # Check shell=True was passed
+        assert call_args.kwargs.get('shell') is True
+
+        # Check the command string contains the double-quoted URL
+        cmd_str = call_args.args[0]
+        assert '"https://mcp.supabase.com/mcp?project_ref=xxx&read_only=true"' in cmd_str
+        assert 'mcp add' in cmd_str
+        assert 'supabase' in cmd_str
+
+    @patch('sys.platform', 'win32')
+    @patch('time.sleep')
+    @patch('subprocess.run')
+    @patch('setup_environment.run_command')
+    @patch('setup_environment.find_command_robust')
+    def test_configure_mcp_server_windows_retry_uses_shell_true(
+        self, mock_find, mock_run_cmd, mock_subprocess_run, mock_sleep,
+    ):
+        """Test Windows retry path uses shell=True with double-quoted URL.
+
+        When both PowerShell and direct execution fail, the retry path should
+        also use subprocess.run with shell=True and double-quoted URL.
+        """
+        assert mock_sleep is not None  # Ensures time.sleep is mocked
+        mock_find.return_value = 'C:\\Users\\Test\\AppData\\Roaming\\npm\\claude.CMD'
+
+        # All run_command calls fail
+        mock_run_cmd.return_value = subprocess.CompletedProcess([], 1, '', 'Failed')
+
+        # First subprocess.run (direct execution) fails, second (retry) succeeds
+        mock_subprocess_run.side_effect = [
+            subprocess.CompletedProcess([], 1, '', 'Direct failed'),
+            subprocess.CompletedProcess([], 0, '', ''),  # Retry succeeds
+        ]
+
+        server = {
+            'name': 'supabase',
+            'scope': 'user',
+            'transport': 'http',
+            'url': 'https://mcp.supabase.com/mcp?project_ref=xxx&read_only=true',
+            'header': 'X-Custom: value',
+        }
+
+        with patch('platform.system', return_value='Windows'):
+            result = setup_environment.configure_mcp_server(server)
+
+        assert result is True
+
+        # Verify subprocess.run was called twice (direct + retry)
+        assert mock_subprocess_run.call_count == 2
+
+        # Check both calls used shell=True and had double-quoted URL
+        for call in mock_subprocess_run.call_args_list:
+            assert call.kwargs.get('shell') is True
+            cmd_str = call.args[0]
+            assert '"https://mcp.supabase.com/mcp?project_ref=xxx&read_only=true"' in cmd_str
+            assert '--header' in cmd_str
+
 
 class TestCreateAdditionalSettings:
     """Test additional settings creation."""

tests/test_setup_environment_additional.py

@@ -618,20 +618,31 @@ def test_configure_mcp_server_missing_transport_details(self, mock_find):
         result = setup_environment.configure_mcp_server(server)
         assert result is False
 
+    @patch('sys.platform', 'win32')
     @patch('platform.system', return_value='Windows')
-    @patch('setup_environment.find_command_robust', return_value='claude')
+    @patch('time.sleep')
+    @patch('subprocess.run')
     @patch('setup_environment.run_command')
-    def test_configure_mcp_server_windows_retry(self, mock_run, mock_find, _mock_system):
+    @patch('setup_environment.find_command_robust', return_value='claude')
+    def test_configure_mcp_server_windows_retry(
+        self, mock_find, mock_run_cmd, mock_subprocess_run, mock_sleep, _mock_system,
+    ):
         """Test MCP configuration retry on Windows."""
         del mock_find  # Unused but required for patch
         del _mock_system  # Unused but required for patch
-        # 3 removes (one per scope), first add attempt fails, retry add succeeds
-        mock_run.side_effect = [
+        del mock_sleep  # Unused but required for patch
+        # run_command: 3 removes (fail) + 1 PowerShell add (fails)
+        mock_run_cmd.side_effect = [
             subprocess.CompletedProcess([], 1, '', 'Server not found'),  # remove user fails
             subprocess.CompletedProcess([], 1, '', 'Server not found'),  # remove local fails
             subprocess.CompletedProcess([], 1, '', 'Server not found'),  # remove project fails
-            subprocess.CompletedProcess([], 1, '', 'Error'),  # first add fails
-            subprocess.CompletedProcess([], 0, '', ''),  # retry add succeeds
+            subprocess.CompletedProcess([], 1, '', 'PowerShell failed'),  # PowerShell add fails
+        ]
+
+        # subprocess.run (shell=True): direct fallback fails, retry succeeds
+        mock_subprocess_run.side_effect = [
+            subprocess.CompletedProcess([], 1, '', 'Direct failed'),  # direct fallback fails
+            subprocess.CompletedProcess([], 0, '', ''),  # retry succeeds
         ]
 
         server = {
@@ -641,11 +652,11 @@ def test_configure_mcp_server_windows_retry(self, mock_run, mock_find, _mock_sys
             'header': 'Auth: token',
         }
 
-        with patch('time.sleep'):  # Skip actual sleep
-            result = setup_environment.configure_mcp_server(server)
+        result = setup_environment.configure_mcp_server(server)
 
         assert result is True
-        assert mock_run.call_count == 5
+        assert mock_run_cmd.call_count == 4  # 3 removes + 1 PowerShell
+        assert mock_subprocess_run.call_count == 2  # direct fallback + retry
 
     @patch('setup_environment.find_command_robust', return_value='claude')
     @patch('setup_environment.run_command')