From c02acab9411f2cc5ea11dd3d771c036ada4aed33 Mon Sep 17 00:00:00 2001 From: Aleksandr Filippov Date: Wed, 7 Jan 2026 21:12:56 +0200 Subject: [PATCH] fix: use shell=True for Windows CMD URL escaping Replace caret (^) escaping approach with shell=True and double-quoted URLs for Windows CMD compatibility. The previous escape_for_cmd() approach failed because subprocess.run() with list arguments uses subprocess.list2cmdline() which follows MSVC quoting rules, not CMD escaping rules. When calling .CMD batch files, Windows automatically invokes cmd.exe /c, causing & characters to be interpreted as command separators despite caret escaping. Using shell=True with double-quoted URLs ensures & characters are protected from CMD interpretation, as documented in Python's subprocess documentation for batch files with special characters. --- scripts/setup_environment.py | 61 ++++----- tests/test_setup_environment.py | 145 ++++++++++++++------- tests/test_setup_environment_additional.py | 29 +++-- 3 files changed, 143 insertions(+), 92 deletions(-) diff --git a/scripts/setup_environment.py b/scripts/setup_environment.py index b10b343..f66fec3 100644 --- a/scripts/setup_environment.py +++ b/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 diff --git a/tests/test_setup_environment.py b/tests/test_setup_environment.py index 37c254a..44b19a8 100644 --- a/tests/test_setup_environment.py +++ b/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¶m2=value2' - escaped = setup_environment.escape_for_cmd(url) - assert escaped == 'https://example.com?param1=value1^¶m2=value2' - - def test_escape_multiple_special_chars(self): - """Test escaping multiple special characters.""" - arg = 'test&value|withchars^and%percent' - escaped = setup_environment.escape_for_cmd(arg) - assert escaped == 'test^&value^|with^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.""" diff --git a/tests/test_setup_environment_additional.py b/tests/test_setup_environment_additional.py index ab78e6f..32e2e39 100644 --- a/tests/test_setup_environment_additional.py +++ b/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')