diff --git a/.github/workflows/compliance.yml b/.github/workflows/compliance.yml index 8fd2eb5..6445449 100644 --- a/.github/workflows/compliance.yml +++ b/.github/workflows/compliance.yml @@ -2,9 +2,9 @@ name: compliance on: push: - branches: [main, "recover-*", "release/**", "hotfix/**"] + branches: [main, "recover-*"] pull_request: - branches: [main, "release/**"] + branches: [main] workflow_dispatch: workflow_call: @@ -123,4 +123,6 @@ jobs: run: | call "%VCVARSALL%" x64 where cl.exe - python -m unittest tests.compliance.test_windows_msvc_smoke + python -m unittest ^ + tests.compliance.test_windows_msvc_smoke ^ + tests.test_windows_pwsh diff --git a/CHANGELOG.md b/CHANGELOG.md index a8cb1ca..9baa2f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +### Changed + +- Windows string commands now prefer a trusted PowerShell 7 launcher and + automatically retain `cmd.exe` compatibility when `pwsh` is unavailable. + The selected shell is visible to agents, and safe mode conservatively gates + dynamic syntax for both interpreters. + ## 0.3.0 - 2026-08-13 ### Changed diff --git a/README.md b/README.md index 685e6a0..4f4a7fe 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,22 @@ walkthroughs, and troubleshooting live in [docs/quickstart.md](docs/quickstart.md) and [docs/mcp-client-config.md](docs/mcp-client-config.md). +On Windows, string commands prefer PowerShell 7 (`pwsh`) with +`-NoLogo -NoProfile -NonInteractive`. If `pwsh` is unavailable, the server +automatically preserves command execution through a trusted `cmd.exe` +compatibility fallback. `server_info`, `check_exec_environment`, and each +`exec_command` result disclose the selected shell so agents can use the right +syntax. Set `CODING_TOOLS_MCP_PWSH_PATH` to an absolute trusted `pwsh.exe` path +to pin PowerShell; an invalid explicit pin is reported instead of ignored. + +Because PowerShell resolves commands at runtime, `safe` mode on Windows gates +dynamic syntax — variables, splatting, call and dot-source operators, `::` +member access, alias/expression evaluation, and nested shells — behind the +`shell_expansion` and `inline_script` permissions. Literal commands are +unaffected; use `request_permissions` or `trusted` mode for the rest. Under the +`cmd.exe` fallback, percent expansion, caret escaping, and `CALL`/`FOR` +evaluation likewise require `shell_expansion`. + ## Seven things to try **1. Make Claude Desktop your coding agent.** The config above is all it @@ -119,9 +135,11 @@ Per-workspace profiles, server and tunnel start/stop, credential setup with clipboard helpers, live health checks. English and 简体中文. **6. Keep an interactive command alive.** `exec_command` starts a REPL or -debugger under a real PTY; `write_stdin` feeds it across turns; `read_output` -pages long output; `kill_command` cleans up. Long-running processes are -first-class, with deadline watchdogs and bounded buffers. +debugger under a real POSIX PTY; `write_stdin` feeds it across turns; +`read_output` pages long output; `kill_command` cleans up. Long-running +processes are first-class, with deadline watchdogs and bounded buffers. +Windows prefers PowerShell 7 and otherwise uses the disclosed `cmd.exe` +fallback for non-TTY commands; ConPTY remains a separate limitation. **7. Give your own agent production-grade hands.** Building an agent loop with the Anthropic SDK or anything else? Don't hand-roll file and exec tools — diff --git a/README.zh-CN.md b/README.zh-CN.md index 6d39621..f5b948b 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -67,6 +67,13 @@ npx coding-tools-mcp --stdio --workspace /path/to/repo # Node 工具链 [docs/quickstart.md](docs/quickstart.md) 与 [docs/mcp-client-config.md](docs/mcp-client-config.md)。 +在 Windows 上，字符串命令优先使用 PowerShell 7（`pwsh`）；未安装时会自动回退到 +可信系统路径中的 `cmd.exe`，不会让 agent 因缺少额外 shell 而失去命令执行能力。 +`server_info`、`check_exec_environment` 和每次 `exec_command` 都会说明实际使用的 +shell，便于 agent 选择正确语法。运维方也可以通过 +`CODING_TOOLS_MCP_PWSH_PATH` 固定一个绝对的 `pwsh.exe` 路径；显式路径配置错误时 +会直接报错，不会掩盖配置问题。 + ## 七个值得一试的玩法 **1. 让 Claude Desktop 成为你的编程 agent。** diff --git a/coding_tools_mcp/processes.py b/coding_tools_mcp/processes.py index 6302b9e..5c28b4a 100644 --- a/coding_tools_mcp/processes.py +++ b/coding_tools_mcp/processes.py @@ -1,12 +1,15 @@ from __future__ import annotations +import functools +import ntpath import os import signal import subprocess import threading import time +from collections.abc import Mapping from dataclasses import dataclass, field -from typing import Any, BinaryIO +from typing import Any, BinaryIO, Literal from .errors import ToolFailure from .textutils import DEFAULT_MAX_LINES, TextTruncation, truncate_text_tail @@ -19,6 +22,287 @@ # agent runtimes. COMMAND_HEAD_BUFFER_DIVISOR = 8 HARD_KILL_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM) +PWSH_PATH_ENV = "CODING_TOOLS_MCP_PWSH_PATH" +CMD_FALLBACK_WARNING = ( + "PowerShell 7 (pwsh) is unavailable; exec_command is using the cmd.exe " + "compatibility fallback. Commands use cmd.exe syntax, not PowerShell syntax. " + "Install PowerShell 7 or set CODING_TOOLS_MCP_PWSH_PATH to use PowerShell." +) + + +@dataclass(frozen=True) +class WindowsCommandShell: + """Trusted Windows command interpreter selected by the server.""" + + kind: Literal["pwsh", "cmd"] + executable: str + fallback: bool = False + fallback_reason: str | None = None + warning: str | None = None + + +def _environment_value(env: Mapping[str, str], target: str) -> str | None: + target_upper = target.upper() + for name, value in env.items(): + if name.upper() == target_upper: + return value + return None + + +def _pwsh_probe_environment() -> dict[str, str]: + """Build a minimal host-derived environment without server secrets.""" + + allowed = ("PATH", "PATHEXT", "SYSTEMROOT", "WINDIR", "TEMP", "TMP", "COMSPEC") + result: dict[str, str] = {} + for name in allowed: + value = _environment_value(os.environ, name) + if value is not None: + result[name] = value + return result + + +def _find_windows_executable_on_path(filename: str, path: str | None) -> str | None: + """Search absolute PATH entries without Windows' implicit current-directory lookup.""" + + if not path: + return None + current_dir = ntpath.normcase(ntpath.abspath(os.getcwd())) + for raw_entry in path.split(";"): + entry = ntpath.expandvars(raw_entry.strip().strip('"')) + if not entry or not ntpath.isabs(entry): + continue + normalized_entry = ntpath.normcase(ntpath.abspath(entry)) + try: + if ntpath.commonpath((current_dir, normalized_entry)) == current_dir: + continue + except ValueError: + pass + candidate = ntpath.normpath(ntpath.join(entry, filename)) + if os.path.isfile(candidate): + return candidate + return None + + +def resolve_pwsh() -> str: + """Resolve PowerShell 7 only from the trusted server process environment.""" + + executable: str | None + configured = (_environment_value(os.environ, PWSH_PATH_ENV) or "").strip().strip('"') + if configured: + executable = ntpath.normpath(ntpath.expandvars(configured)) + if ( + not ntpath.isabs(executable) + or ntpath.basename(executable).lower() != "pwsh.exe" + or not os.path.isfile(executable) + ): + raise ToolFailure( + "SHELL_NOT_FOUND", + f"{PWSH_PATH_ENV} must point to an existing absolute pwsh.exe path.", + category="runtime", + details={"executable": executable, "environment_variable": PWSH_PATH_ENV}, + ) + else: + executable = _find_windows_executable_on_path( + "pwsh.exe", + _environment_value(os.environ, "PATH"), + ) + if executable is None: + raise ToolFailure( + "SHELL_NOT_FOUND", + "PowerShell 7 was not found on the server process PATH.", + category="runtime", + details={ + "executable": "pwsh", + "retry_hint": ( + "Install PowerShell 7 and add pwsh to PATH, or set " + f"{PWSH_PATH_ENV} to its absolute path." + ), + }, + ) + major = pwsh_major_version(executable) + if major < 7: + raise ToolFailure( + "SHELL_VERSION_UNSUPPORTED", + f"PowerShell 7 or newer is required; resolved major version {major}.", + category="runtime", + details={"executable": executable, "major_version": major, "required_major_version": 7}, + ) + return executable + + +def resolve_cmd() -> str: + """Resolve cmd.exe from trusted server-process Windows locations.""" + + raw_candidates: list[str] = [] + comspec = (_environment_value(os.environ, "COMSPEC") or "").strip().strip('"') + if comspec: + raw_candidates.append(comspec) + for variable in ("SYSTEMROOT", "WINDIR"): + root = (_environment_value(os.environ, variable) or "").strip().strip('"') + if root: + raw_candidates.append(ntpath.join(root, "System32", "cmd.exe")) + + seen: set[str] = set() + for raw_candidate in raw_candidates: + candidate = ntpath.normpath(ntpath.expandvars(raw_candidate)) + normalized = ntpath.normcase(candidate) + if normalized in seen: + continue + seen.add(normalized) + if ( + ntpath.isabs(candidate) + and ntpath.basename(candidate).lower() == "cmd.exe" + and os.path.isfile(candidate) + ): + return candidate + + executable = _find_windows_executable_on_path( + "cmd.exe", + _environment_value(os.environ, "PATH"), + ) + if executable is not None: + return executable + raise ToolFailure( + "SHELL_NOT_FOUND", + "Neither PowerShell 7 nor cmd.exe could be resolved for Windows string commands.", + category="runtime", + details={ + "executable": "cmd.exe", + "retry_hint": ( + "Restore the Windows command processor or install PowerShell 7 and add " + f"pwsh to PATH (or set {PWSH_PATH_ENV})." + ), + }, + ) + + +def resolve_windows_command_shell() -> WindowsCommandShell: + """Prefer PowerShell 7 and automatically retain cmd.exe compatibility.""" + + configured_pwsh = bool((_environment_value(os.environ, PWSH_PATH_ENV) or "").strip()) + try: + return WindowsCommandShell(kind="pwsh", executable=resolve_pwsh()) + except ToolFailure as exc: + # An explicit pin is operator intent. Falling back would hide a typo or + # silently ignore a version constraint the operator expected to hold. + if configured_pwsh: + raise + return WindowsCommandShell( + kind="cmd", + executable=resolve_cmd(), + fallback=True, + fallback_reason=exc.code, + warning=CMD_FALLBACK_WARNING, + ) + + +_selected_shell_lock = threading.Lock() +_selected_shell: WindowsCommandShell | ToolFailure | None = None + + +def selected_windows_command_shell(*, refresh: bool = False) -> WindowsCommandShell: + """Resolve the Windows shell once per process and pin the answer. + + Re-resolving per call would let a transient probe failure flip one exec to + cmd.exe while the next uses pwsh, and would let concurrent requests each + start their own multi-second version probe before command accounting sees + them. The lock serializes the first resolution; every later call reuses the + pinned selection (including a pinned failure) until ``refresh`` — used by + check_exec_environment so operators can pick up a newly installed shell + without restarting the server. + """ + + global _selected_shell + with _selected_shell_lock: + if refresh or _selected_shell is None: + try: + _selected_shell = resolve_windows_command_shell() + except ToolFailure as failure: + _selected_shell = failure + pinned = _selected_shell + if isinstance(pinned, ToolFailure): + raise pinned + return pinned + + +def _reset_selected_windows_command_shell() -> None: + """Drop the pinned shell selection (test seam).""" + + global _selected_shell + with _selected_shell_lock: + _selected_shell = None + + +@functools.lru_cache(maxsize=16) +def pwsh_major_version(executable: str) -> int: + """Return and cache the major version reported by a pwsh executable.""" + + try: + completed = subprocess.run( + [ + executable, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + "$PSVersionTable.PSVersion.Major", + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=5, + check=False, + env=_pwsh_probe_environment(), + cwd=ntpath.dirname(executable) or None, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise ToolFailure( + "SHELL_VERSION_UNSUPPORTED", + "PowerShell version could not be verified.", + category="runtime", + details={"executable": executable, "reason": str(exc)}, + ) from exc + stdout = completed.stdout.strip() + if completed.returncode != 0 or not stdout.isdigit(): + raise ToolFailure( + "SHELL_VERSION_UNSUPPORTED", + "PowerShell version could not be verified.", + category="runtime", + details={ + "executable": executable, + "exit_code": completed.returncode, + "stderr": completed.stderr.strip()[:500], + }, + ) + return int(stdout) + + +def build_pwsh_argv(executable: str, command: str) -> list[str]: + """Build a deterministic non-interactive PowerShell invocation.""" + + return [ + executable, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + command, + ] + + +def build_cmd_command_line(executable: str, command: str) -> str: + """Build the exact cmd.exe command line, bypassing MS CRT argv quoting. + + Passing an argument list would make Popen serialize it with list2cmdline, + whose ``\\"`` escapes reach cmd.exe verbatim and corrupt any command that + contains quotes. ``/S /C`` is specified against a raw command line instead: + cmd strips the first and last quote and executes everything between them + unchanged, so the payload needs exactly one enclosing quote pair and no + further escaping. + """ + + return f'"{executable}" /D /V:OFF /S /C "{command}"' def terminate_process_group( @@ -69,10 +353,18 @@ def spawn_process( env: dict[str, str], tty: bool, popen_kwargs: dict[str, Any], + windows_shell: WindowsCommandShell | None = None, ) -> tuple[subprocess.Popen[bytes], int | None]: """Spawn a pipe-backed or true POSIX PTY-backed process.""" if not tty: + if os.name == "nt" and shell and isinstance(command, str): + selected_shell = windows_shell or selected_windows_command_shell() + if selected_shell.kind == "pwsh": + command = build_pwsh_argv(selected_shell.executable, command) + else: + command = build_cmd_command_line(selected_shell.executable, command) + shell = False process = subprocess.Popen( command, cwd=cwd, diff --git a/coding_tools_mcp/server.py b/coding_tools_mcp/server.py index 7f31bfc..ffaaffe 100644 --- a/coding_tools_mcp/server.py +++ b/coding_tools_mcp/server.py @@ -62,6 +62,8 @@ COMMAND_BUFFER_BYTES, COMMAND_HEAD_BUFFER_DIVISOR, CommandRun, + WindowsCommandShell, + selected_windows_command_shell, spawn_process, start_reader_threads, start_command_watchdog, @@ -187,11 +189,64 @@ class ModeCapabilities: r"(https?://|urllib\.request|urllib3|requests\.|http\.client|\bHTTPConnection\b|\bHTTPSConnection\b|socket\.|aiohttp|httpx|\bcurl\b|\bwget\b|\bnc\b|\bnetcat\b|\bssh\b|\bscp\b|\bftp\b)", re.I, ) +POWERSHELL_NETWORK_RE = re.compile( + r"(?:^|[;&|{}\r\n])\s*(?:[\w.]+[\\/])?(?:Invoke-WebRequest|Invoke-RestMethod|Start-BitsTransfer|" + r"Test-NetConnection|Test-Connection|Resolve-DnsName|iwr|irm|tnc|ping(?:\.exe)?|" + r"nslookup(?:\.exe)?|tracert(?:\.exe)?)\b|\b(?:System\.)?Net\.", + re.I, +) SHELL_EXPANSION_RE = re.compile(r"(`|\$\(|\$\{)") DESTRUCTIVE_RE = re.compile( r"(^|\s)(sudo|su|chmod\s+-R|chown\s+-R|mkfs|mount|umount|find\b[^;&|]*\s-delete\b|git\b[^;&|]*\breset\s+--hard\b|git\b[^;&|]*\bclean\s+-[^\s]*[fx][^\s]*|rm\s+-[^\s]*r[^\s]*f|rm\s+-[^\s]*f[^\s]*r)\b", re.I, ) +# A module-qualified spelling (Microsoft.PowerShell.Management\Remove-Item) +# invokes the same cmdlet, so the command-position match accepts an optional +# Module\ prefix in both PowerShell scans above and below. +POWERSHELL_DESTRUCTIVE_RE = re.compile( + r"(?:^|[;&|{}\r\n])\s*(?:[\w.]+[\\/])?(?:Remove-Item|rm|ri|del|erase|rmdir|rd)\b" + r"(?=[^;&|{}\r\n]*\s-(?:r|re|rec|recu|recur|recurs|recurse)\b)", + re.I, +) +# Single-quoted PowerShell strings are inert text ('' is the only escape), so +# the dynamic-syntax scan drops them before matching; a literal '$5' or 'a::b' +# argument is not expansion. Backticks never reach this far in safe mode +# because SHELL_EXPANSION_RE already gates them. +POWERSHELL_SINGLE_QUOTED_RE = re.compile(r"'[^']*(?:''[^']*)*'") +# PowerShell resolves commands at runtime, so scanning for cmdlet names cannot +# see through a variable, a splatted parameter set, a redefined alias, or a +# .NET member call. Any of those constructs makes the destructive and network +# scans above unsound, so they require the same explicit permission that POSIX +# command substitution already requires instead of being scanned for keywords. +POWERSHELL_DYNAMIC_RE = re.compile( + r"(?P\$)" + r"|(?P(?:^|[\s;&|(){},=])@)" + r"|(?P::)" + r"|(?P(?:^|[;|(){}\r\n]|&&|\|\|)\s*(?:&(?!&)|\.)\s)" + r"|(?P\b(?:Invoke-Expression|iex|Invoke-Command|icm|New-Object|" + r"Add-Type|Set-Alias|New-Alias|sal|nal)\b)", + re.I, +) +# cmd.exe also resolves command text at runtime. Paired %...% expansion, +# CALL/FOR double evaluation, and caret escaping can hide the literal command +# from the scans above, so safe mode treats those constructs like shell +# expansion. A lone % cannot expand on a command line (echo 100%, +# git log --format=%h), and CALL/FOR only evaluate at command position +# (echo call for help stays literal), so neither is gated. +CMD_DYNAMIC_RE = re.compile( + r"(?P%[^%\r\n]+%)" + r"|(?P\^)" + r"|(?P(?:^|[&|()\r\n])\s*(?:call|for)\b)", + re.I, +) +CMD_DESTRUCTIVE_RE = re.compile( + r"(?:^|[&|()\r\n])\s*(?:(?:del|erase|rd|rmdir)\b" + r"(?=[^&|()\r\n]*\s/(?:s|s[q]?|q[s]))" + # format.com and diskpart are real executables, so a path spelling such as + # C:\Windows\System32\format.com invokes them just as well as the bare name. + r"|(?:[^\s&|()\r\n]*[\\/])?(?:format(?:\.com)?|diskpart)(?:\.exe)?(?=$|[\s&|()<>\r\n]))", + re.I, +) MAX_HTTP_REQUEST_BYTES = 1_048_576 EXEC_PREVIEW_BYTES = 4096 MAX_ACTIVE_COMMANDS = 16 @@ -1545,8 +1600,8 @@ def git_path_filter(self, raw_path: str) -> str: return "." return self.resolve_for_write(raw_path).display - def _exec_environment_summary(self) -> dict[str, Any]: - return { + def _exec_environment_summary(self, *, refresh_command_shell: bool = False) -> dict[str, Any]: + summary: dict[str, Any] = { "workspace": str(self.workspace.root), "permission_mode": self.permission_mode, "network_allowed": self.allow_network, @@ -1555,6 +1610,18 @@ def _exec_environment_summary(self) -> dict[str, Any]: "tmpdir": str(self.command_tmp_dir()), "cache_dir": str(self.cache_dir), } + if os.name == "nt": + try: + summary["command_shell"] = windows_command_shell_payload( + selected_windows_command_shell(refresh=refresh_command_shell) + ) + except ToolFailure as exc: + summary["command_shell"] = { + "available": False, + "error_code": exc.code, + "message": exc.message, + } + return summary def _landlock_enforced(self, landlock: dict[str, Any]) -> bool: return bool(landlock.get("available")) and self.landlock_enabled() @@ -1668,6 +1735,10 @@ def server_info(self, args: dict[str, Any]) -> dict[str, Any]: def check_exec_environment(self, args: dict[str, Any]) -> dict[str, Any]: landlock = landlock_status_payload() + # The explicit diagnostic re-resolves the pinned Windows shell so an + # operator who just installed pwsh sees (and activates) it here without + # restarting the server. + summary = self._exec_environment_summary(refresh_command_shell=True) warnings: list[str] = [] if not landlock.get("available"): warnings.append("Linux Landlock filesystem confinement is unavailable") @@ -1677,9 +1748,16 @@ def check_exec_environment(self, args: dict[str, Any]) -> dict[str, Any]: warnings.append( "tools/list annotations are faked as read-only; apply_patch and exec_command still mutate and execute" ) + command_shell = summary.get("command_shell") + if isinstance(command_shell, dict): + shell_warning = command_shell.get("warning") + if isinstance(shell_warning, str): + warnings.append(shell_warning) + elif command_shell.get("available") is False: + warnings.append(str(command_shell.get("message") or "Windows command shell is unavailable")) return { "ok": True, - **self._exec_environment_summary(), + **summary, "landlock_enabled": self._landlock_enforced(landlock), "landlock_abi": landlock.get("abi_version"), "global_tmp_write": self.global_tmp_write_policy(), @@ -2357,11 +2435,12 @@ def exec_command(self, args: dict[str, Any]) -> dict[str, Any]: workdir = self.resolve_existing(str(workdir_arg)) if not workdir.path.is_dir(): raise ToolFailure("NOT_A_DIRECTORY", "workdir is not a directory.", category="validation") - self._check_command_policy(cmd, args) + tty = bool(args.get("tty", False)) + windows_shell = selected_windows_command_shell() if os.name == "nt" and not tty else None + self._check_command_policy(cmd, args, windows_shell=windows_shell) timeout_ms = int(args.get("timeout_ms", 30000)) yield_ms = int(args.get("yield_time_ms", 10000)) max_output_bytes = int(args.get("max_output_bytes", 65536)) - tty = bool(args.get("tty", False)) stdin_text = str(args.get("stdin", "")) env = self._command_env(args.get("env", {})) start = time.time() @@ -2413,11 +2492,20 @@ def exec_command(self, args: dict[str, Any]) -> dict[str, Any]: env=env, tty=tty, popen_kwargs=popen_extra, + windows_shell=windows_shell, ) + command_warnings = [ + warning + for warning in ( + landlock_warning, + windows_shell.warning if windows_shell is not None else None, + ) + if warning + ] command = self._make_command( process, timeout_at=deadline, - warnings=[landlock_warning] if landlock_warning else None, + warnings=command_warnings, pty_master_fd=pty_master_fd, ) with self.commands_lock: @@ -2460,6 +2548,8 @@ def finish() -> dict[str, Any]: # terminated/timeout) so exec, polling, and kill paths agree. payload = command.snapshot_since_cursor(max_output_bytes) payload["elapsed_ms"] = int((time.time() - start) * 1000) + if windows_shell is not None: + payload["command_shell"] = windows_command_shell_payload(windows_shell) self._add_exec_diagnostics(payload) return self._format_command_output(command, payload, args) @@ -2484,10 +2574,30 @@ def finish() -> dict[str, Any]: return finish() time.sleep(0.02) - def _check_command_policy(self, cmd: str, args: dict[str, Any]) -> None: + def _check_command_policy( + self, + cmd: str, + args: dict[str, Any], + *, + windows_shell: WindowsCommandShell | None = None, + ) -> None: if self.dangerously_skip_all_permissions: return - self._check_command_paths(cmd) + compact = " ".join(cmd.split()).lower() + # POSIX shlex treats backslash as an escape and silently eats it, which + # blinds every token-based check to unquoted Windows paths such as + # C:\Windows\System32\...\powershell.exe. Backslash is a path separator + # on both Windows shells (their escapes are backtick and caret), so the + # token scans run on a slash-normalized copy there. + scan_cmd = cmd.replace("\\", "/") if windows_shell is not None else cmd + if windows_shell is not None and windows_shell.kind == "cmd" and CMD_DESTRUCTIVE_RE.search(cmd): + raise ToolFailure( + "PERMISSION_REQUIRED", + "Destructive commands are blocked without explicit permission.", + category="permission", + details={"permission": "destructive_command", "command": compact}, + ) + self._check_command_paths(scan_cmd) env = args.get("env", {}) if isinstance(env, dict) and any( is_filtered_env_var(str(key), str(value)) for key, value in env.items() @@ -2499,7 +2609,7 @@ def _check_command_policy(self, cmd: str, args: dict[str, Any]) -> None: details={"permission": "sensitive_env", "env_keys": sorted(str(key) for key in env)}, ) if not self.capabilities.inline_script: - inline_script = inline_script_command(cmd) + inline_script = inline_script_command(scan_cmd) if inline_script is not None: raise ToolFailure( "PERMISSION_REQUIRED", @@ -2507,7 +2617,12 @@ def _check_command_policy(self, cmd: str, args: dict[str, Any]) -> None: category="permission", details={"permission": INLINE_SCRIPT_PERMISSION, **inline_script}, ) - compact = " ".join(cmd.split()).lower() + uses_powershell = powershell_executes_string_commands(windows_shell) + uses_cmd = windows_shell is not None and windows_shell.kind == "cmd" + # The PowerShell scans model PowerShell semantics; running them against + # a POSIX shell or cmd.exe command would gate rm -r and ping on hosts + # where those regexes were never the contract. + scan_powershell = uses_powershell if not self.capabilities.shell_expansion and SHELL_EXPANSION_RE.search(cmd): raise ToolFailure( "PERMISSION_REQUIRED", @@ -2522,20 +2637,58 @@ def _check_command_policy(self, cmd: str, args: dict[str, Any]) -> None: category="permission", details={"permission": "destructive_command", "command": compact}, ) - if DESTRUCTIVE_RE.search(cmd): + if ( + DESTRUCTIVE_RE.search(cmd) + or (scan_powershell and POWERSHELL_DESTRUCTIVE_RE.search(cmd)) + ): raise ToolFailure( "PERMISSION_REQUIRED", "Destructive commands are blocked without explicit permission.", category="permission", details={"permission": "destructive_command", "command": compact}, ) - if not self.allow_network and NETWORK_RE.search(cmd) and not is_literal_network_reference_command(cmd): + network_command = NETWORK_RE.search(cmd) or ( + scan_powershell and POWERSHELL_NETWORK_RE.search(cmd) + ) + if not self.allow_network and network_command and not is_literal_network_reference_command(scan_cmd): raise ToolFailure( "PERMISSION_REQUIRED", "Network access is denied by default.", category="permission", details={"permission": "network", "command": compact}, ) + # Runs last so a command the scans above already recognized keeps its + # precise permission label; this is the catch-all for the PowerShell + # syntax that makes those scans unsound in the first place. + if not self.capabilities.shell_expansion and uses_powershell: + construct = powershell_dynamic_construct(cmd) + if construct is not None: + raise ToolFailure( + "PERMISSION_REQUIRED", + "PowerShell dynamic syntax requires explicit permission because the command " + "a variable, splat, alias, or .NET member resolves to cannot be verified " + "statically.", + category="permission", + details={ + "permission": "shell_expansion", + "construct": construct, + "command": compact, + }, + ) + if not self.capabilities.shell_expansion and uses_cmd: + construct = cmd_dynamic_construct(cmd) + if construct is not None: + raise ToolFailure( + "PERMISSION_REQUIRED", + "cmd.exe dynamic syntax requires explicit permission because expansion, " + "double evaluation, and escaping can hide the command from policy scans.", + category="permission", + details={ + "permission": "shell_expansion", + "construct": construct, + "command": compact, + }, + ) def _add_exec_diagnostics(self, payload: dict[str, Any]) -> None: diagnostics = exec_output_diagnostics(payload) @@ -3766,6 +3919,20 @@ def inline_script_segment(command: str | None, args: list[str]) -> dict[str, str return {"command": name, "option": option} if name in {"ruby", "perl"} and "-e" in args: return {"command": name, "option": "-e"} + if name in {"pwsh", "pwsh.exe", "powershell", "powershell.exe"}: + for arg in args: + if not arg.startswith("-"): + continue + option = arg.lstrip("-").lower() + # PowerShell accepts any unambiguous prefix, so -e, -enc, and + # -EncodedCommand all smuggle a base64 script past text scanning. + if option and ("command".startswith(option) or "encodedcommand".startswith(option)): + return {"command": name, "option": arg} + return None + if name in {"cmd", "cmd.exe"}: + for arg in args: + if arg.lstrip("-/").lower() in {"c", "k"}: + return {"command": name, "option": arg} return None @@ -3920,6 +4087,50 @@ def is_inspectable_path_argument(token: str) -> bool: return "." in PurePosixPath(normalized).name +def windows_command_shell_payload(shell: WindowsCommandShell) -> dict[str, Any]: + payload: dict[str, Any] = { + "available": True, + "kind": shell.kind, + "executable": shell.executable, + "fallback": shell.fallback, + } + if shell.fallback_reason is not None: + payload["fallback_reason"] = shell.fallback_reason + if shell.warning is not None: + payload["warning"] = shell.warning + return payload + + +def powershell_executes_string_commands( + windows_shell: WindowsCommandShell | None = None, +) -> bool: + """True when the selected Windows string-command shell is PowerShell 7. + + Command policy has to mirror the selection passed to processes.spawn_process: + PowerShell syntax is only worth gating where PowerShell is the interpreter. + The host fallback keeps direct policy-helper tests backward compatible. + """ + + if windows_shell is not None: + return windows_shell.kind == "pwsh" + return os.name == "nt" + + +def powershell_dynamic_construct(command: str) -> str | None: + scannable = POWERSHELL_SINGLE_QUOTED_RE.sub("''", command) + match = POWERSHELL_DYNAMIC_RE.search(scannable) + if match is None: + return None + return match.lastgroup + + +def cmd_dynamic_construct(command: str) -> str | None: + match = CMD_DYNAMIC_RE.search(command) + if match is None: + return None + return match.lastgroup + + def is_literal_network_reference_command(command: str) -> bool: try: tokens = shlex_split(command) diff --git a/docs/ci-and-tests.md b/docs/ci-and-tests.md index 6e632a7..dd32ee0 100644 --- a/docs/ci-and-tests.md +++ b/docs/ci-and-tests.md @@ -116,6 +116,8 @@ Windows reports unsupported TTY requests explicitly, force-kills a background command without relying on POSIX `SIGKILL`, initializes Visual Studio with `vcvarsall.bat x64`, checks the narrow default `core` environment, and confirms that `--shell-env-inherit all` can compile and run a single-file `cl.exe` smoke. +It also exercises PowerShell 7 selection, the trusted `cmd.exe` compatibility +fallback, and the shell-specific safe-mode policy gates. Manual SWE-bench workflow: diff --git a/docs/limitations.md b/docs/limitations.md index 9bc981a..0170fb2 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -6,6 +6,28 @@ - Non-Linux platforms or Linux kernels without Landlock are not production targets for `exec_command` without an external sandbox. - This build uses real POSIX PTYs but does not implement Windows ConPTY; `tty=true` returns `TTY_UNSUPPORTED` on Windows. +- Windows string commands prefer PowerShell 7 (`pwsh`) and run it with + `-NoLogo -NoProfile -NonInteractive`. Operators may pin an absolute trusted + executable with `CODING_TOOLS_MCP_PWSH_PATH`; otherwise the server searches + absolute entries on its own process `PATH` while excluding the current + directory tree. If no usable unpinned `pwsh` is found, the server + automatically falls back to a trusted `cmd.exe` resolved from the server + environment or Windows system directory. An invalid explicit PowerShell pin + remains an error. +- Windows `safe` mode cannot statically decide what PowerShell dynamic syntax + resolves to, so variables (`$`), splatting (`@`), the call and dot-source + operators, .NET member access (`::`), and alias or expression evaluation + cmdlets require the `shell_expansion` permission even when the command would + turn out to be harmless. Nested shells (`pwsh -Command`, `pwsh + -EncodedCommand`, `cmd /c`) require `inline_script`. Use + `request_permissions` or `trusted` mode for commands that need them. Command + scanning is not a sandbox: this build has no OS-level confinement on Windows, + so `safe` mode there is a best-effort gate rather than a boundary. +- The `cmd.exe` compatibility fallback disables AutoRun and delayed expansion. + In `safe` mode, percent expansion, caret escaping, and `CALL`/`FOR` evaluation + require `shell_expansion`; recursive `del`/`rmdir`, `format`, and `diskpart` + remain permission-gated. These checks are also best-effort rather than an + OS-level boundary. - Portable filesystems do not provide a transaction across unrelated directories. `apply_patch` keeps same-directory backups and rolls back the full staged set, but a storage failure that also prevents rollback is surfaced diff --git a/docs/runtime-contract-v0.3.md b/docs/runtime-contract-v0.3.md index 5e7005b..eabb0fd 100644 --- a/docs/runtime-contract-v0.3.md +++ b/docs/runtime-contract-v0.3.md @@ -293,7 +293,7 @@ Retry: This command_id has expired or never existed; … Known tool error codes include: ```json -["ABSOLUTE_PATH_DENIED", "BINARY_FILE", "COMMAND_CLOSED", "COMMAND_LIMIT_REACHED", "COMMAND_NOT_FOUND", "ELICITATION_UNSUPPORTED", "GIT_ERROR", "INTERNAL_ERROR", "INVALID_ARGUMENT", "IS_DIRECTORY", "NOT_A_DIRECTORY", "NOT_FOUND", "OUTPUT_TOO_LARGE", "PATCH_CONFLICT", "PATCH_CONTEXT_AMBIGUOUS", "PATCH_CONTEXT_NOT_FOUND", "PATCH_FAILED", "PATCH_HUNKS_OVERLAP", "PATCH_ROLLBACK_FAILED", "PATH_OUTSIDE_WORKSPACE", "PERMISSION_REQUIRED", "RUNTIME_DIR_UNWRITABLE", "SANDBOX_UNAVAILABLE", "SYMLINK_ESCAPE", "TTY_UNSUPPORTED", "UNSUPPORTED_ENCODING"] +["ABSOLUTE_PATH_DENIED", "BINARY_FILE", "COMMAND_CLOSED", "COMMAND_LIMIT_REACHED", "COMMAND_NOT_FOUND", "ELICITATION_UNSUPPORTED", "GIT_ERROR", "INTERNAL_ERROR", "INVALID_ARGUMENT", "IS_DIRECTORY", "NOT_A_DIRECTORY", "NOT_FOUND", "OUTPUT_TOO_LARGE", "PATCH_CONFLICT", "PATCH_CONTEXT_AMBIGUOUS", "PATCH_CONTEXT_NOT_FOUND", "PATCH_FAILED", "PATCH_HUNKS_OVERLAP", "PATCH_ROLLBACK_FAILED", "PATH_OUTSIDE_WORKSPACE", "PERMISSION_REQUIRED", "RUNTIME_DIR_UNWRITABLE", "SANDBOX_UNAVAILABLE", "SHELL_NOT_FOUND", "SHELL_VERSION_UNSUPPORTED", "SYMLINK_ESCAPE", "TTY_UNSUPPORTED", "UNSUPPORTED_ENCODING"] ``` Error categories are `validation`, `security`, `permission`, `runtime`, @@ -336,6 +336,13 @@ bounded, all of them per workspace rather than per client. Completed commands have a TTL. POSIX `tty=true` uses a real pseudo-terminal; Windows reports `TTY_UNSUPPORTED` in this build instead of pretending pipes are a TTY. +Windows string commands prefer a server-resolved PowerShell 7 (`pwsh`) and +automatically use a trusted `cmd.exe` compatibility fallback when no usable +unpinned PowerShell is available. An invalid +`CODING_TOOLS_MCP_PWSH_PATH` remains an error. Windows `exec_command` results +include a structured `command_shell`; fallback results also include a warning +so an agent can use cmd syntax rather than retrying PowerShell syntax. + ## HTTP authentication Non-loopback deployment requires bearer or OAuth authentication unless the @@ -375,7 +382,8 @@ Annotations: `{"title":"Server info","readOnlyHint":true,"destructiveHint":false Returns server version, `supported_protocol_versions`, workspace, fixed tool count, auth state, permission mode, runtime directories, project-context -metadata, exec policy, and the static retained-output budget. It reports no +metadata, exec policy, the selected Windows `command_shell` when applicable, +and the static retained-output budget. It reports no per-session value and no runtime counter: there is no session, and how often a budget was hit is a property of the process rather than an answer to whichever client asked. Those counters travel with telemetry. @@ -386,7 +394,8 @@ Inputs: none. Annotations: `{"title":"Check exec environment","readOnlyHint":true,"destructiveHint":false,"idempotentHint":true,"openWorldHint":false}`. -Returns lightweight policy and Landlock status without running active probes. +Returns lightweight policy, Landlock status, and the selected Windows command +shell. A `cmd.exe` fallback is repeated in `warnings`. ### read_file diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index f0bc4c9..f724d56 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -22,6 +22,26 @@ If `exec_command` returns a warning about Linux Landlock being unavailable, the If an older client or server reports `SANDBOX_UNAVAILABLE` as an error, upgrade to the current behavior or run on a Landlock-capable Linux kernel. +## Windows Command Shell + +Windows string commands prefer PowerShell 7 (`pwsh`). If it is not installed or +an unpinned copy cannot be verified, the server automatically uses the trusted +Windows `cmd.exe` compatibility fallback. The selected interpreter appears as +`command_shell` in `server_info`, `check_exec_environment`, and +`exec_command`; fallback results also include a warning so the agent can switch +to cmd syntax. + +The selection is resolved once and pinned for the server process, so concurrent +commands always agree on one interpreter. `check_exec_environment` re-resolves +the pin: after installing PowerShell 7, run that tool (or restart the server) to +leave the `cmd.exe` fallback. + +To require a specific PowerShell installation, set +`CODING_TOOLS_MCP_PWSH_PATH` to its absolute `pwsh.exe` path. A bad explicit pin +returns `SHELL_NOT_FOUND` or `SHELL_VERSION_UNSUPPORTED` instead of falling back. +If neither interpreter can be resolved, restore `cmd.exe` or install PowerShell +7. + ## Command Hangs Or Times Out If the result returns `status: "running"`, poll with `write_stdin` using empty `chars`, or terminate with `kill_command`. Command deadlines still apply when the client stops polling. diff --git a/tests/compliance/test_windows_msvc_smoke.py b/tests/compliance/test_windows_msvc_smoke.py index 6e40269..9a93ac5 100644 --- a/tests/compliance/test_windows_msvc_smoke.py +++ b/tests/compliance/test_windows_msvc_smoke.py @@ -132,7 +132,10 @@ def test_inherit_all_preserves_msvc_environment_for_single_file_compile(self) -> run_result = client.call_tool( "exec_command", { - "cmd": "hello.exe", + # PowerShell intentionally does not search the current + # directory for executables; this spelling also works + # unchanged under the cmd.exe compatibility fallback. + "cmd": r".\hello.exe", "timeout_ms": 30000, "yield_time_ms": 30000, "max_output_bytes": 20000, diff --git a/tests/test_windows_pwsh.py b/tests/test_windows_pwsh.py new file mode 100644 index 0000000..dae98f8 --- /dev/null +++ b/tests/test_windows_pwsh.py @@ -0,0 +1,664 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from coding_tools_mcp import processes +from coding_tools_mcp import server +from coding_tools_mcp.errors import ToolFailure +from coding_tools_mcp.server import Runtime, ShellEnvPolicy + +PWSH_SHELL = processes.WindowsCommandShell( + kind="pwsh", + executable=r"C:\Program Files\PowerShell\7\pwsh.exe", +) +CMD_SHELL = processes.WindowsCommandShell( + kind="cmd", + executable=r"C:\Windows\System32\cmd.exe", + fallback=True, +) + + +class WindowsPowerShellSpawnTests(unittest.TestCase): + def tearDown(self) -> None: + processes.pwsh_major_version.cache_clear() + processes._reset_selected_windows_command_shell() + + def test_windows_string_command_uses_server_resolved_noninteractive_pwsh(self) -> None: + captured: dict[str, object] = {} + + class FakeProcess: + pass + + def fake_popen(command: object, **kwargs: object) -> FakeProcess: + captured["command"] = command + captured.update(kwargs) + return FakeProcess() + + trusted = r"C:\Program Files\PowerShell\7\pwsh.exe" + trusted_path = r"C:\Program Files\PowerShell\7;C:\Windows\System32" + attacker_path = r"C:\workspace\bin" + + with ( + patch.object(processes.os, "name", "nt"), + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict(processes.os.environ, {"Path": trusted_path}, clear=True), + patch.object(processes.os.path, "isfile", side_effect=lambda path: path == trusted), + patch.object(processes, "pwsh_major_version", return_value=7), + patch.object(processes.subprocess, "Popen", side_effect=fake_popen), + ): + process, pty_fd = processes.spawn_process( + "Write-Output 'ok'", + cwd=r"C:\workspace", + shell=True, + env={"Path": attacker_path}, + tty=False, + popen_kwargs={}, + ) + + self.assertIsInstance(process, FakeProcess) + self.assertIsNone(pty_fd) + self.assertEqual( + captured["command"], + [ + trusted, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + "Write-Output 'ok'", + ], + ) + self.assertIs(captured["shell"], False) + self.assertEqual(captured["env"], {"Path": attacker_path}) + + def test_windows_string_command_falls_back_to_trusted_cmd(self) -> None: + captured: dict[str, object] = {} + + class FakeProcess: + pass + + def fake_popen(command: object, **kwargs: object) -> FakeProcess: + captured["command"] = command + captured.update(kwargs) + return FakeProcess() + + trusted = r"C:\Windows\System32\cmd.exe" + with ( + patch.object(processes.os, "name", "nt"), + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict( + processes.os.environ, + { + "COMSPEC": trusted, + "Path": r"C:\Windows\System32", + "SystemRoot": r"C:\Windows", + }, + clear=True, + ), + patch.object(processes.os.path, "isfile", side_effect=lambda path: path == trusted), + patch.object(processes.subprocess, "Popen", side_effect=fake_popen), + ): + process, pty_fd = processes.spawn_process( + "echo ok", + cwd=r"C:\workspace", + shell=True, + env={"Path": r"C:\workspace\bin"}, + tty=False, + popen_kwargs={}, + ) + + self.assertIsInstance(process, FakeProcess) + self.assertIsNone(pty_fd) + self.assertEqual( + captured["command"], + f'"{trusted}" /D /V:OFF /S /C "echo ok"', + ) + self.assertIs(captured["shell"], False) + self.assertEqual(captured["env"], {"Path": r"C:\workspace\bin"}) + + def test_cmd_command_line_preserves_embedded_quotes_verbatim(self) -> None: + # An argv list would be re-serialized with MS CRT rules and reach + # cmd.exe as \" sequences, corrupting any quoted argument. The raw + # /S /C command line must carry the payload byte-for-byte. + command = 'type "C:\\Program Files\\notes.txt" > "out dir\\copy.txt"' + line = processes.build_cmd_command_line(r"C:\Windows\System32\cmd.exe", command) + self.assertEqual( + line, + '"C:\\Windows\\System32\\cmd.exe" /D /V:OFF /S /C ' + '"type "C:\\Program Files\\notes.txt" > "out dir\\copy.txt""', + ) + + def test_windows_pwsh_resolution_skips_current_directory_entries(self) -> None: + trusted = r"C:\Program Files\PowerShell\7\pwsh.exe" + malicious = r"C:\workspace\pwsh.exe" + + with ( + patch.object(processes.os, "name", "nt"), + patch.object(processes.os, "getcwd", return_value=r"C:\workspace"), + patch.dict( + processes.os.environ, + {"Path": r".;C:\workspace;C:\Program Files\PowerShell\7"}, + clear=True, + ), + patch.object( + processes.os.path, + "isfile", + side_effect=lambda path: path in {trusted, malicious}, + ), + patch.object(processes, "pwsh_major_version", return_value=7), + ): + self.assertEqual(processes.resolve_pwsh(), trusted) + + def test_windows_pwsh_rejects_invalid_explicit_path(self) -> None: + with ( + patch.object(processes.os, "name", "nt"), + patch.dict( + processes.os.environ, + {processes.PWSH_PATH_ENV: r".\pwsh.exe"}, + clear=True, + ), + patch.object(processes.os.path, "isfile", return_value=True), + ): + with self.assertRaises(ToolFailure) as raised: + processes.resolve_pwsh() + + self.assertEqual(raised.exception.code, "SHELL_NOT_FOUND") + + def test_invalid_explicit_pwsh_pin_does_not_fall_back(self) -> None: + with ( + patch.dict( + processes.os.environ, + {processes.PWSH_PATH_ENV: r".\pwsh.exe"}, + clear=True, + ), + patch.object(processes.os.path, "isfile", return_value=False), + patch.object(processes, "resolve_cmd") as resolve_cmd, + ): + with self.assertRaises(ToolFailure) as raised: + processes.resolve_windows_command_shell() + + self.assertEqual(raised.exception.code, "SHELL_NOT_FOUND") + resolve_cmd.assert_not_called() + + def test_windows_shell_falls_back_when_pwsh_is_missing(self) -> None: + cmd = r"C:\Windows\System32\cmd.exe" + with ( + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict( + processes.os.environ, + { + "COMSPEC": cmd, + "Path": r"C:\Windows\System32", + "SystemRoot": r"C:\Windows", + }, + clear=True, + ), + patch.object(processes.os.path, "isfile", side_effect=lambda path: path == cmd), + ): + selected = processes.resolve_windows_command_shell() + + self.assertEqual(selected.kind, "cmd") + self.assertEqual(selected.executable, cmd) + self.assertTrue(selected.fallback) + self.assertEqual(selected.fallback_reason, "SHELL_NOT_FOUND") + self.assertIn("cmd.exe syntax", selected.warning or "") + + def test_windows_string_command_rejects_powershell_older_than_7(self) -> None: + executable = r"C:\PowerShell\6\pwsh.exe" + with ( + patch.object(processes.os, "name", "nt"), + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict(processes.os.environ, {"Path": r"C:\PowerShell\6"}, clear=True), + patch.object(processes.os.path, "isfile", side_effect=lambda path: path == executable), + patch.object(processes, "pwsh_major_version", return_value=6), + ): + with self.assertRaises(ToolFailure) as raised: + processes.resolve_pwsh() + + self.assertEqual(raised.exception.code, "SHELL_VERSION_UNSUPPORTED") + + def test_unpinned_unsupported_pwsh_falls_back_to_cmd(self) -> None: + pwsh = r"C:\PowerShell\6\pwsh.exe" + cmd = r"C:\Windows\System32\cmd.exe" + with ( + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict( + processes.os.environ, + { + "COMSPEC": cmd, + "Path": r"C:\PowerShell\6;C:\Windows\System32", + }, + clear=True, + ), + patch.object( + processes.os.path, + "isfile", + side_effect=lambda path: path in {pwsh, cmd}, + ), + patch.object(processes, "pwsh_major_version", return_value=6), + ): + selected = processes.resolve_windows_command_shell() + + self.assertEqual(selected.kind, "cmd") + self.assertEqual(selected.fallback_reason, "SHELL_VERSION_UNSUPPORTED") + + def test_windows_shell_reports_error_when_neither_interpreter_exists(self) -> None: + with ( + patch.object(processes.os, "getcwd", return_value=r"C:\server"), + patch.dict(processes.os.environ, {"Path": r"C:\empty"}, clear=True), + patch.object(processes.os.path, "isfile", return_value=False), + ): + with self.assertRaises(ToolFailure) as raised: + processes.resolve_windows_command_shell() + + self.assertEqual(raised.exception.code, "SHELL_NOT_FOUND") + + @unittest.skipUnless(os.name == "nt", "requires Windows PowerShell 7") + def test_runtime_executes_bare_powershell_syntax(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime( + Path(tmp), + permission_mode="trusted", + shell_env_policy=ShellEnvPolicy(inherit="all"), + ) + try: + result = runtime.exec_command( + { + "cmd": "Write-Output ('PS_MAJOR=' + $PSVersionTable.PSVersion.Major)", + "timeout_ms": 30_000, + "yield_time_ms": 30_000, + "verbosity": "full", + } + ) + finally: + runtime.close() + + self.assertEqual(result.get("status"), "exited", result) + self.assertEqual(result.get("exit_code"), 0, result) + self.assertIn("PS_MAJOR=7", str(result.get("stdout", ""))) + + @unittest.skipUnless(os.name == "nt", "requires Windows cmd.exe") + def test_runtime_cmd_fallback_is_visible_and_executes(self) -> None: + cmd = os.environ["COMSPEC"] + selected = processes.WindowsCommandShell( + kind="cmd", + executable=cmd, + fallback=True, + fallback_reason="SHELL_NOT_FOUND", + warning=processes.CMD_FALLBACK_WARNING, + ) + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime( + Path(tmp), + permission_mode="trusted", + shell_env_policy=ShellEnvPolicy(inherit="all"), + ) + try: + with patch.object( + server, "selected_windows_command_shell", return_value=selected + ): + result = runtime.exec_command( + { + "cmd": "echo CMD_FALLBACK_OK", + "timeout_ms": 30_000, + "yield_time_ms": 30_000, + "verbosity": "full", + } + ) + finally: + runtime.close() + + self.assertEqual(result.get("exit_code"), 0, result) + self.assertIn("CMD_FALLBACK_OK", str(result.get("stdout", ""))) + self.assertEqual(result.get("command_shell", {}).get("kind"), "cmd") + self.assertTrue(result.get("command_shell", {}).get("fallback")) + self.assertTrue( + any("cmd.exe compatibility fallback" in warning for warning in result.get("warnings", [])), + result, + ) + + @unittest.skipUnless(os.name == "nt", "requires Windows cmd.exe") + def test_runtime_real_cmd_fallback_without_pwsh_on_path(self) -> None: + # No mocked resolver or Popen: hide every PATH entry that offers + # pwsh.exe, let selection fall back to the real cmd.exe, and run a + # command whose quoted argument would be corrupted by argv re-quoting. + def offers_pwsh(entry: str) -> bool: + candidate = entry.strip().strip('"') + if not candidate: + return False + try: + return (Path(candidate) / "pwsh.exe").is_file() + except OSError: + return False + + stripped_path = ";".join( + entry for entry in os.environ.get("PATH", "").split(";") if not offers_pwsh(entry) + ) + with tempfile.TemporaryDirectory() as tmp: + (Path(tmp) / "has space.txt").write_text("FALLBACK_QUOTES_OK", encoding="utf-8") + runtime = Runtime( + Path(tmp), + permission_mode="trusted", + shell_env_policy=ShellEnvPolicy(inherit="all"), + ) + try: + with patch.dict(os.environ, {"PATH": stripped_path}, clear=False): + os.environ.pop(processes.PWSH_PATH_ENV, None) + processes._reset_selected_windows_command_shell() + result = runtime.exec_command( + { + "cmd": 'type "has space.txt"', + "timeout_ms": 30_000, + "yield_time_ms": 30_000, + "verbosity": "full", + } + ) + finally: + processes._reset_selected_windows_command_shell() + runtime.close() + + self.assertEqual(result.get("exit_code"), 0, result) + self.assertIn("FALLBACK_QUOTES_OK", str(result.get("stdout", ""))) + self.assertEqual(result.get("command_shell", {}).get("kind"), "cmd") + self.assertTrue(result.get("command_shell", {}).get("fallback"), result) + + def test_server_info_and_exec_check_disclose_cmd_fallback(self) -> None: + selected = processes.WindowsCommandShell( + kind="cmd", + executable=r"C:\Windows\System32\cmd.exe", + fallback=True, + fallback_reason="SHELL_NOT_FOUND", + warning=processes.CMD_FALLBACK_WARNING, + ) + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp)) + try: + summary = {"command_shell": server.windows_command_shell_payload(selected)} + with patch.object(runtime, "_exec_environment_summary", return_value=summary): + info = runtime.server_info_payload() + check = runtime.check_exec_environment({}) + finally: + runtime.close() + + self.assertEqual(info.get("command_shell", {}).get("kind"), "cmd") + self.assertTrue(info.get("command_shell", {}).get("fallback")) + self.assertTrue( + any("cmd.exe compatibility fallback" in warning for warning in check.get("warnings", [])), + check, + ) + + +class WindowsShellSelectionCacheTests(unittest.TestCase): + def tearDown(self) -> None: + processes._reset_selected_windows_command_shell() + + def test_selection_is_pinned_for_the_process(self) -> None: + with patch.object( + processes, "resolve_windows_command_shell", return_value=PWSH_SHELL + ) as resolver: + first = processes.selected_windows_command_shell() + second = processes.selected_windows_command_shell() + + self.assertIs(first, PWSH_SHELL) + self.assertIs(second, PWSH_SHELL) + resolver.assert_called_once() + + def test_selection_failure_is_pinned_until_refresh(self) -> None: + failure = ToolFailure("SHELL_NOT_FOUND", "no shell", category="runtime") + with patch.object( + processes, + "resolve_windows_command_shell", + side_effect=[failure, CMD_SHELL], + ) as resolver: + with self.assertRaises(ToolFailure): + processes.selected_windows_command_shell() + # A pinned failure must not re-probe on the exec path. + with self.assertRaises(ToolFailure): + processes.selected_windows_command_shell() + self.assertEqual(resolver.call_count, 1) + refreshed = processes.selected_windows_command_shell(refresh=True) + + self.assertIs(refreshed, CMD_SHELL) + self.assertEqual(resolver.call_count, 2) + + +class WindowsPowerShellPolicyTests(unittest.TestCase): + def test_safe_mode_blocks_powershell_network_commands_without_url_literals(self) -> None: + commands = ( + "Invoke-WebRequest -Uri example.com", + "Invoke-RestMethod -Uri api.example.com", + "iwr example.com", + "irm api.example.com", + "Start-BitsTransfer -Source example.com -Destination out.bin", + "New-Object System.Net.WebClient", + "Test-NetConnection example.com", + "Test-Connection example.com", + "ping example.com", + "tnc example.com", + "Resolve-DnsName example.com", + "[System.Net.Dns]::GetHostAddresses('example.com')", + "Write-Output ok\nInvoke-WebRequest -Uri example.com", + # Module-qualified invocation runs the same cmdlet. + "Microsoft.PowerShell.Utility\\Invoke-WebRequest -Uri example.com", + ) + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp), permission_mode="safe") + try: + for command in commands: + with self.subTest(command=command): + with self.assertRaises(ToolFailure) as raised: + runtime._check_command_policy(command, {}, windows_shell=PWSH_SHELL) + self.assertEqual(raised.exception.details.get("permission"), "network") + finally: + runtime.close() + + def test_safe_mode_blocks_recursive_powershell_deletion_aliases_and_abbreviations(self) -> None: + commands = ( + "Remove-Item -Recurse .", + "Remove-Item -Recurse -Force .", + "Remove-Item -Force -LiteralPath . -Recurse", + "Remove-Item -Recu -For .", + "rm -r .", + "rm -Recurse -Force .", + "ri -Recu .", + "ri -Force -Recurse .", + "del -Recurse .", + "rmdir -r .", + "Write-Output ok\r\nRemove-Item -Recurse .", + # Module-qualified invocation runs the same cmdlet. + "Microsoft.PowerShell.Management\\Remove-Item -Recurse -Force build", + ) + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp), permission_mode="safe") + try: + for command in commands: + with self.subTest(command=command): + with self.assertRaises(ToolFailure) as raised: + runtime._check_command_policy(command, {}, windows_shell=PWSH_SHELL) + self.assertEqual( + raised.exception.details.get("permission"), + "destructive_command", + ) + finally: + runtime.close() + + +class WindowsPowerShellDynamicSyntaxTests(unittest.TestCase): + """PowerShell resolves commands at runtime, so cmdlet-name scanning alone + cannot decide whether a command is destructive or reaches the network.""" + + def assert_policy(self, command: str, *, mode: str = "safe") -> ToolFailure | None: + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp), permission_mode=mode) + try: + try: + runtime._check_command_policy(command, {}, windows_shell=PWSH_SHELL) + except ToolFailure as failure: + return failure + finally: + runtime.close() + return None + + def test_safe_mode_gates_command_names_built_from_variables_and_splatting(self) -> None: + # Both of these pass every cmdlet-name scan while still invoking a + # network download and a recursive delete. + cases = { + "$c='Invoke-WebRequest'; & $c example.com": "expansion", + "$p=@{Recurse=$true}; Remove-Item . @p": "expansion", + "Set-Alias grab Invoke-WebRequest; grab example.com": "dynamic_eval", + "[IO.File]::Delete('C:\\data\\report.csv')": "static_member", + "Remove-Item . @args": "splatting", + ". .\\payload.ps1": "call_operator", + "iex (Get-Content payload.txt -Raw)": "dynamic_eval", + } + for command, construct in cases.items(): + with self.subTest(command=command): + failure = self.assert_policy(command) + self.assertIsNotNone(failure, f"{command!r} was allowed in safe mode") + assert failure is not None + self.assertEqual(failure.details.get("permission"), "shell_expansion") + self.assertEqual(failure.details.get("construct"), construct) + + def test_safe_mode_gates_nested_shells_that_smuggle_encoded_scripts(self) -> None: + for command in ( + "pwsh -EncodedCommand SQBuAHYAbwBrAGUA", + "pwsh -enc SQBuAHYAbwBrAGUA", + "pwsh -Command Invoke-WebRequest example.com", + "powershell.exe -c whoami", + "cmd /c del /s /q C:\\data", + # Unquoted absolute interpreter paths must not slip past the gate + # just because POSIX tokenization eats backslashes. + "C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe -EncodedCommand SQBuAHYAbwBrAGUA", + "C:\\Windows\\System32\\cmd.exe /c del /s /q data", + ): + with self.subTest(command=command): + failure = self.assert_policy(command) + self.assertIsNotNone(failure, f"{command!r} was allowed in safe mode") + assert failure is not None + self.assertEqual(failure.details.get("permission"), "inline_script") + + def test_safe_mode_still_allows_literal_powershell_commands(self) -> None: + for command in ( + "Get-ChildItem -Path src -Recurse -Name", + "Write-Output ok", + "git commit -m 'fix parser' && git log --oneline -1", + "git config user.email dev@example.com", + ".\\build.exe --release", + "Get-Content README.md | Select-String planet.txt", + # Single-quoted PowerShell strings are inert: no expansion happens. + "Write-Output '$5'", + "git commit -m 'refs #12 :: costs $5'", + ): + with self.subTest(command=command): + self.assertIsNone(self.assert_policy(command), f"{command!r} was blocked in safe mode") + + def test_trusted_mode_allows_dynamic_syntax(self) -> None: + for command in ("$c='Get-Date'; & $c", "Remove-Item . @p"): + with self.subTest(command=command): + self.assertIsNone(self.assert_policy(command, mode="trusted")) + + def test_network_scan_does_not_flag_words_ending_in_net(self) -> None: + self.assertIsNone(server.POWERSHELL_NETWORK_RE.search("Get-Content planet.txt")) + self.assertIsNotNone(server.POWERSHELL_NETWORK_RE.search("New-Object System.Net.WebClient")) + self.assertIsNotNone(server.POWERSHELL_NETWORK_RE.search("[Net.Dns]::GetHostAddresses('a')")) + + +class PosixPolicyIsolationTests(unittest.TestCase): + """Without a Windows shell the PowerShell and cmd.exe scans must stay off: + they would otherwise rewrite the POSIX policy contract (rm -r without -f + and ping were never gated there).""" + + def assert_policy(self, command: str) -> ToolFailure | None: + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp), permission_mode="safe") + try: + try: + runtime._check_command_policy(command, {}) + except ToolFailure as failure: + return failure + finally: + runtime.close() + return None + + def test_posix_hosts_keep_their_existing_expansion_policy(self) -> None: + self.assertIsNone(self.assert_policy("echo $HOME")) + + def test_posix_hosts_do_not_inherit_powershell_scans(self) -> None: + for command in ("rm -r build", "ping example.com", "Remove-Item -Recurse ."): + with self.subTest(command=command): + self.assertIsNone( + self.assert_policy(command), + f"{command!r} was blocked by a Windows-only scan on POSIX", + ) + + +class WindowsCmdPolicyTests(unittest.TestCase): + def assert_policy(self, command: str, *, mode: str = "safe") -> ToolFailure | None: + with tempfile.TemporaryDirectory() as tmp: + runtime = Runtime(Path(tmp), permission_mode=mode) + try: + try: + runtime._check_command_policy(command, {}, windows_shell=CMD_SHELL) + except ToolFailure as failure: + return failure + finally: + runtime.close() + return None + + def test_safe_mode_blocks_recursive_cmd_deletion(self) -> None: + for command in ( + "del /s /q data", + "rmdir /s /q build", + "format.com D:", + "format D:", + "diskpart", + # Path spellings invoke the same executables. + "C:\\Windows\\System32\\format.com D:", + ): + with self.subTest(command=command): + failure = self.assert_policy(command) + self.assertIsNotNone(failure) + assert failure is not None + self.assertEqual(failure.details.get("permission"), "destructive_command") + + def test_safe_mode_gates_cmd_dynamic_syntax(self) -> None: + cases = { + "%COMSPEC% /c echo ok": "expansion", + "echo %PATH%": "expansion", + "c^u^r^l example.com": "escape", + "call tool.cmd": "dynamic_eval", + "for %f in (*.txt) do type %f": "dynamic_eval", + } + for command, construct in cases.items(): + with self.subTest(command=command): + failure = self.assert_policy(command) + self.assertIsNotNone(failure) + assert failure is not None + self.assertEqual(failure.details.get("permission"), "shell_expansion") + self.assertEqual(failure.details.get("construct"), construct) + + def test_safe_mode_allows_literal_cmd_commands(self) -> None: + for command in ( + "echo ok", + "git status", + "hello.exe", + "cl.exe /nologo hello.c", + # A lone % cannot expand on a cmd command line. + "git log --format=%h", + "echo 100%", + # CALL and FOR only evaluate at command position. + "echo call for help", + ): + with self.subTest(command=command): + self.assertIsNone(self.assert_policy(command), f"{command!r} was blocked in safe mode") + + def test_trusted_mode_allows_cmd_dynamic_syntax(self) -> None: + self.assertIsNone(self.assert_policy("%COMSPEC% /c echo ok", mode="trusted")) + + +if __name__ == "__main__": + unittest.main()