diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index faaec2d5..2bfd4703 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -30,6 +30,14 @@ _DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024 # 1MB buffer limit MINIMUM_CLAUDE_CODE_VERSION = "2.0.0" +# cmd.exe metacharacters (plus the quote character cmd.exe uses to toggle +# its quoting state, and "!", which expands like "%" when delayed expansion +# is enabled). subprocess.list2cmdline quotes arguments for the MSVCRT argv +# rules only -- it adds quotes only around whitespace -- so in a +# whitespace-free argument these characters reach a cmd.exe command line +# verbatim. See _reject_windows_batch_cli / _reject_windows_cmd_metacharacters. +_CMD_EXE_METACHARACTERS = '&|<>^%!"' + # Track live CLI subprocesses so we can terminate them when the parent Python # process exits. This mirrors the TypeScript SDK's parent-exit cleanup and # prevents orphaned `claude` processes from leaking when callers crash or exit @@ -147,22 +155,75 @@ def _find_cli(self) -> str: return bundled_cli # Fall back to system-wide search + which_hit: str | None = None if cli := shutil.which("claude"): - return cli - - locations = [ - Path.home() / ".npm-global/bin/claude", - Path("/usr/local/bin/claude"), - Path.home() / ".local/bin/claude", - Path.home() / "node_modules/.bin/claude", - Path.home() / ".yarn/bin/claude", - Path.home() / ".claude/local/claude", - ] + if platform.system() != "Windows" or self._is_windows_native_exe(cli): + return cli + # Windows resolved something CreateProcess cannot run directly + # as the CLI: npm's claude.cmd shim (which connect() refuses to + # spawn) or an extensionless wrapper script from a git-bash / + # WSL setup (which fails at spawn with WinError 193). shutil.which + # walks PATH directory-major, so such an entry in an early PATH + # directory shadows a native claude.exe installed in a later + # one (within one directory the default PATHEXT would prefer + # .EXE, so the shadowing is purely the directory order). Prefer + # any discoverable native executable, and keep this hit only as + # the last resort so a shim-only machine still gets the + # explanatory batch-script refusal from connect(). The claude.exe + # probe is vetted too: PATHEXT resolution can append an + # extension and hand back "claude.exe.cmd". + exe = shutil.which("claude.exe") + if exe and self._is_windows_native_exe(exe): + return exe + which_hit = cli + + if platform.system() == "Windows": + # Only the native installer's claude.exe. Path.exists() does + # no PATHEXT resolution, so the .exe name must be probed + # explicitly. The POSIX-shaped entries below are deliberately + # not probed on Windows: an extensionless match (a WSL / git-bash + # script artifact at ~/.local/bin/claude) would preempt the + # explanatory batch-script refusal with an opaque spawn + # failure, and a rooted-but-driveless "/usr/local/bin/claude" + # resolves against the current drive (C:\usr\local\bin\...), + # a location another local user can create -- a + # binary-planting probe. + locations = [Path.home() / ".local/bin/claude.exe"] + else: + locations = [ + Path.home() / ".npm-global/bin/claude", + Path("/usr/local/bin/claude"), + Path.home() / ".local/bin/claude", + Path.home() / "node_modules/.bin/claude", + Path.home() / ".yarn/bin/claude", + Path.home() / ".claude/local/claude", + ] for path in locations: if path.exists() and path.is_file(): return str(path) + if which_hit is not None: + # No native executable was discoverable anywhere: return the + # original which() hit so connect() raises the batch-script + # refusal (with its remediation) for a shim, or the spawn error + # for a wrapper script, rather than a bare not-found error. + return which_hit + + if platform.system() == "Windows": + # npm's Windows install is a claude.cmd shim, which connect() + # refuses (_reject_windows_batch_cli), so do not recommend it. + raise CLINotFoundError( + "Claude Code not found. Install the native claude.exe with " + "(PowerShell):\n" + " irm https://claude.ai/install.ps1 | iex\n" + "\nOr install the claude-agent-sdk wheel for a platform that " + "bundles claude.exe (e.g. Windows x64), or provide the path to " + "a claude.exe via ClaudeAgentOptions:\n" + " ClaudeAgentOptions(cli_path='C:\\\\path\\\\to\\\\claude.exe')\n" + "\n(npm install -g @anthropic-ai/claude-code produces a claude.cmd " + "shim, which this SDK refuses to run on Windows.)" + ) raise CLINotFoundError( "Claude Code not found. Install with:\n" " npm install -g @anthropic-ai/claude-code\n" @@ -187,6 +248,122 @@ def _find_bundled_cli(self) -> str | None: return None + @staticmethod + def _is_windows_native_exe(cli_path: str) -> bool: + """Whether cli_path's final component names an image CreateProcess + runs directly (.exe / .com), used only to decide which discovery + result to prefer. It is not a security gate: every returned path + still passes _reject_windows_batch_cli in connect(). + """ + name = cli_path.replace("\\", "/").rsplit("/", 1)[-1] + return name.rstrip(". ").lower().endswith((".exe", ".com")) + + @staticmethod + def _is_windows_batch_cli(cli_path: str) -> bool: + """Whether cli_path names a .bat/.cmd batch script on Windows. + + Always False off Windows. See _reject_windows_batch_cli for why + spawning such a script is refused. + """ + if platform.system() != "Windows": + return False + # Deliberately NOT pathlib: PureWindowsPath and PurePosixPath parse + # several of the cases below differently (".cmd" has suffix ".cmd" + # on POSIX but "" on Windows), and the tests run on POSIX CI while + # the code runs on Windows. Plain string logic behaves identically + # on both. + # + # Classify EVERY path component, not only the final one. Win32 opens + # a path after lexical normalization -- "." / ".." collapsing, + # repeated separators, and position-dependent trailing dot/space + # trimming (a middle ".. " or "..." stays a literal name while a + # final one trims to ".." or vanishes) -- and any attempt to + # re-derive the effective final component here is a race against + # that ruleset: get one rule slightly wrong and a spelling such as + # "claude.cmd\\...\\.." resolves to claude.cmd on Windows while the + # simulation lands on some other name. Refusing whenever ANY + # component carries a batch extension closes that whole class + # outright, because every normalization trick still has to spell + # the .bat/.cmd component somewhere in the string. It costs + # nothing legitimate: no real claude.exe lives beneath a directory + # named like a batch file. + # + # Within a component, Win32 finds the extension with a last-dot + # scan over the WHOLE component, stream spec included -- + # "claude:evil.cmd" has extension ".cmd" -- while an NTFS stream + # spec also opens its base file -- "claude.cmd:stream" opens + # claude.cmd -- and a drive prefix ("C:claude.cmd") rides in the + # same component. Splitting each component on ":" covers all of + # these: colons cannot appear in real file names, so no legitimate + # segment is over-refused. Trailing dots and spaces, which Windows + # strips at path resolution, are stripped per segment (the same + # normalization Rust's CVE-2024-24576 fix applies), and a bare + # ".cmd" counts as a batch extension (as Win32 PathFindExtension + # treats it, and pathlib does not). + return any( + segment.rstrip(". ").lower().endswith((".bat", ".cmd")) + for component in cli_path.replace("\\", "/").split("/") + for segment in component.split(":") + ) + + @staticmethod + def _reject_windows_batch_cli(cli_path: str) -> None: + """Refuse to execute a .bat/.cmd script as the CLI on Windows. + + Windows has no shebang mechanism: CreateProcess runs batch scripts + by silently rewriting the spawn into a 'cmd.exe /c' invocation, and + cmd.exe re-parses the whole command line at execution time. + subprocess.list2cmdline quotes arguments for the MSVCRT argv rules + only, not for cmd.exe, so cmd.exe metacharacters inside an argument + value -- for example a session title passed to --resume -- reach + cmd.exe unescaped and can execute injected commands. Reliable + escaping for cmd.exe does not exist (%VAR% expands even inside + double quotes), so spawning a batch script with runtime-provided + arguments cannot be made safe. Refusing is the same remediation + Node.js shipped for this vulnerability class (CVE-2024-27980, + "BatBadBut"). + + In practice this refuses npm's claude.cmd shim, which _find_cli + returns only when no native claude.exe is discoverable (for + example sdist installs on a machine with just the npm shim). The + alternatives in the error message avoid cmd.exe entirely. + """ + if not SubprocessCLITransport._is_windows_batch_cli(cli_path): + return + raise CLIConnectionError( + f"Refusing to execute batch script {cli_path!r}: Windows runs " + ".bat/.cmd files via cmd.exe, which can execute commands " + "injected through CLI arguments, and no reliable escaping for " + "cmd.exe exists. Use a native claude executable instead: " + "install Claude Code natively " + "(irm https://claude.ai/install.ps1 | iex), point " + "ClaudeAgentOptions(cli_path=...) at a claude.exe, or install " + "the claude-agent-sdk wheel for a platform that bundles " + "claude.exe (e.g. Windows x64)." + ) + + @staticmethod + def _reject_windows_cmd_metacharacters(option_name: str, value: str) -> None: + """Defense in depth for Windows: reject cmd.exe metacharacters. + + With batch-script spawning refused (_reject_windows_batch_cli), + these characters are harmless: list2cmdline quotes correctly for + native executables. They are rejected anyway so that resume / + session_id values, which applications commonly take from external + input, stay inert even if a cmd.exe hop is ever reintroduced + between the SDK and the CLI. No format is imposed beyond this + (resume values may be arbitrary session titles, not only UUIDs), + and POSIX behavior is unchanged. + """ + if platform.system() != "Windows": + return + bad = sorted({c for c in value if c in _CMD_EXE_METACHARACTERS or c in "\r\n"}) + if bad: + raise ValueError( + f"{option_name} value {value!r} contains characters that " + f"are unsafe to pass on a Windows command line: {bad!r}" + ) + def _build_settings_value(self) -> str | None: """Build settings value, merging sandbox settings if provided. @@ -355,9 +532,13 @@ def _build_command(self) -> list[str]: # a separate CLI flag -- letting an untrusted value inject arbitrary # flags. The equals form always binds the value to the flag. if self._options.resume: + self._reject_windows_cmd_metacharacters("resume", self._options.resume) cmd.append(f"--resume={self._options.resume}") if self._options.session_id: + self._reject_windows_cmd_metacharacters( + "session_id", self._options.session_id + ) cmd.append(f"--session-id={self._options.session_id}") # Handle settings and sandbox: merge sandbox into settings if both are provided @@ -431,6 +612,14 @@ def _build_command(self) -> list[str]: if value is None: # Boolean flag without value cmd.append(f"--{flag}") + elif str(value).startswith("-"): + # In the two-token form, a dash-leading value is not bound + # to its flag when the CLI declares the option with an + # optional value -- it parses as a separate flag instead + # (the same injection the --resume change above closes). + # The equals form always binds. Mirrors the equivalent + # guard in the TypeScript SDK. + cmd.append(f"--{flag}={value}") else: # Flag with value cmd.extend([f"--{flag}", str(value)]) @@ -483,6 +672,10 @@ async def connect(self) -> None: if self._cli_path is None: self._cli_path = await anyio.to_thread.run_sync(self._find_cli) + # Validate the resolved CLI before anything is spawned with it -- + # this guards the version probe below as well as the main spawn. + self._reject_windows_batch_cli(self._cli_path) + if not os.environ.get("CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK"): await self._check_claude_version() diff --git a/tests/test_transport.py b/tests/test_transport.py index 999dccde..4c1a5abc 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -2258,3 +2258,509 @@ async def _test() -> None: await proc.wait() anyio.run(_test) + + +def _mock_connect_processes() -> tuple[MagicMock, MagicMock]: + """Build the (version probe, main process) mocks connect() awaits.""" + version_process = MagicMock() + version_process.stdout = MagicMock() + version_process.stdout.receive = AsyncMock(return_value=b"2.0.0 (Claude Code)") + version_process.terminate = MagicMock() + version_process.wait = AsyncMock() + + main_process = MagicMock() + main_process.stdout = MagicMock() + main_stdin = MagicMock() + main_stdin.aclose = AsyncMock() + main_process.stdin = main_stdin + main_process.returncode = None + return version_process, main_process + + +class TestWindowsBatchScriptRefusal: + """connect() must never spawn a .bat/.cmd script on Windows. + + CreateProcess routes batch scripts through cmd.exe /c, and cmd.exe + re-parses the whole command line, so every argument value would reach + a shell. The refusal happens before the version probe, so no spawn of + the script occurs at all. + """ + + _PLATFORM = "claude_agent_sdk._internal.transport.subprocess_cli.platform.system" + + def test_npm_cmd_shim_from_which_is_refused(self): + # Shim-only machine: which("claude") resolves npm's claude.cmd and + # no native claude.exe is discoverable (which("claude.exe") -> None, + # no .exe in the fallback locations). Discovery must still hand the + # shim to connect() so the batch-script refusal fires -- the .exe + # preference is additive and never lets a shim-only machine spawn. + async def _test(): + from claude_agent_sdk._errors import CLIConnectionError + + shim = "C:\\Users\\u\\AppData\\Roaming\\npm\\claude.CMD" + + def _which(name: str) -> str | None: + return shim if name == "claude" else None + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + side_effect=_which, + ), + patch("pathlib.Path.exists", return_value=False), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLIConnectionError, match="batch script"), + ): + await transport.connect() + + assert mock_open.call_count == 0 + + anyio.run(_test) + + def test_native_exe_is_preferred_over_shadowing_npm_shim(self): + # Dual-install machine: npm's %APPDATA%\npm precedes the native + # installer's %USERPROFILE%\.local\bin on PATH, so which("claude") + # resolves the claude.cmd shim -- shutil.which walks PATH + # directory-major, so the earlier npm directory wins (within one + # directory the default PATHEXT would prefer .EXE over .CMD; the + # shadowing comes purely from directory order). Discovery must find + # the shadowed native claude.exe via which("claude.exe") so connect() + # proceeds instead of refusing. + async def _test(): + shim = "C:\\Users\\u\\AppData\\Roaming\\npm\\claude.CMD" + native = "C:\\Users\\u\\.local\\bin\\claude.exe" + + def _which(name: str) -> str | None: + return {"claude": shim, "claude.exe": native}.get(name) + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + version_process, main_process = _mock_connect_processes() + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + side_effect=_which, + ), + patch("pathlib.Path.exists", return_value=False), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + ): + mock_open.side_effect = [version_process, main_process] + await transport.connect() + + assert mock_open.call_count == 2 + assert mock_open.call_args_list[1].args[0][0] == native + + anyio.run(_test) + + def test_claude_exe_probe_result_is_vetted(self): + # Python 3.12+ shutil.which appends PATHEXT extensions even to a + # name that already carries one, so which("claude.exe") can hand + # back a stray "claude.exe.cmd". Discovery must not accept that as + # the rescued native exe: it falls through to the fallback location + # and, with none there, returns the original npm shim so connect() + # refuses naming the shim its remediation message is written for. + async def _test(): + from claude_agent_sdk._errors import CLIConnectionError + + shim = "C:\\Users\\u\\AppData\\Roaming\\npm\\claude.CMD" + junk = "C:\\tools\\claude.exe.cmd" + + def _which(name: str) -> str | None: + return {"claude": shim, "claude.exe": junk}.get(name) + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + side_effect=_which, + ), + patch("pathlib.Path.exists", return_value=False), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLIConnectionError, match=r"npm\\\\claude\.CMD"), + ): + await transport.connect() + + assert mock_open.call_count == 0 + + anyio.run(_test) + + def test_extensionless_which_hit_still_prefers_native_exe(self): + # Python 3.12+ shutil.which also probes the bare name, so an + # extensionless git-bash / WSL wrapper script named "claude" in an + # early PATH directory shadows a native claude.exe installed in a + # later one. CreateProcess cannot run that script (WinError 193), + # so discovery must run the same native-exe rescue instead of + # committing to the wrapper. + async def _test(): + wrapper = "C:\\Users\\u\\bin\\claude" + native = "C:\\Users\\u\\.local\\bin\\claude.exe" + + def _which(name: str) -> str | None: + return {"claude": wrapper, "claude.exe": native}.get(name) + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + version_process, main_process = _mock_connect_processes() + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + side_effect=_which, + ), + patch("pathlib.Path.exists", return_value=False), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + ): + mock_open.side_effect = [version_process, main_process] + await transport.connect() + + assert mock_open.call_count == 2 + assert mock_open.call_args_list[1].args[0][0] == native + + anyio.run(_test) + + def test_explicit_bat_cli_path_is_refused(self): + async def _test(): + from claude_agent_sdk._errors import CLIConnectionError + + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeAgentOptions(cli_path="C:\\tools\\claude.bat"), + ) + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLIConnectionError, match="batch script"), + ): + await transport.connect() + + assert mock_open.call_count == 0 + + anyio.run(_test) + + @pytest.mark.parametrize( + "cli_path", + [ + "C:\\tools\\claude.cmd.", + "C:\\tools\\claude.CMD ", + "C:\\tools\\claude.cmd:stream", + "C:\\tools\\.cmd", + "C:claude.cmd", + "C:/tools/claude.cmd", + "\\\\server\\share\\claude.cmd", + "C:\\tools\\claude.cmd\\.", + "C:\\tools\\claude.cmd\\x\\..", + "C:\\tools\\claude.cmd\\x\\.. ", + "C:\\tools\\claude.cmd\\x\\.. .", + "C:\\tools\\claude.cmd\\\\.", + "C:/tools/claude.cmd//x/..", + "C:\\tools\\claude.cmd\\", + "C:\\tools\\claude.cmd\\...", + "C:\\tools\\claude.cmd\\....", + "C:\\tools\\claude:evil.cmd", + "C:\\tools\\claude.exe:evil.cmd", + ":claude.cmd", + # A middle dots/spaces-only component is a literal name on Win32 + # (trailing-dot trimming applies to the final segment only), so + # a following ".." pops that literal and lands on claude.cmd. + "C:\\tools\\claude.cmd\\...\\..", + "C:\\tools\\claude.cmd\\. .\\..", + "C:\\tools\\claude.cmd\\ \\..", + "C:\\tools\\claude.cmd\\.. \\..", + ], + ) + def test_suffix_tricks_are_refused(self, cli_path: str): + async def _test(): + from claude_agent_sdk._errors import CLIConnectionError + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions(cli_path=cli_path) + ) + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLIConnectionError, match="batch script"), + ): + await transport.connect() + + assert mock_open.call_count == 0 + + anyio.run(_test) + + def test_native_exe_is_allowed_on_windows(self): + async def _test(): + transport = SubprocessCLITransport( + prompt="test", + options=ClaudeAgentOptions( + cli_path="C:\\Users\\u\\.local\\bin\\claude.EXE" + ), + ) + version_process, main_process = _mock_connect_processes() + + with ( + patch(self._PLATFORM, return_value="Windows"), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + ): + mock_open.side_effect = [version_process, main_process] + await transport.connect() + + assert mock_open.call_count == 2 + + anyio.run(_test) + + @pytest.mark.parametrize("system", ["Linux", "Darwin"]) + def test_posix_platforms_are_unchanged(self, system: str): + async def _test(): + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + version_process, main_process = _mock_connect_processes() + + with ( + patch(self._PLATFORM, return_value=system), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + return_value="/usr/local/bin/claude", + ) as mock_which, + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + ): + mock_open.side_effect = [version_process, main_process] + await transport.connect() + + # POSIX discovery uses the which("claude") result directly: the + # native-exe preference is a Windows-only branch, so there is no + # claude.exe probe here. + assert mock_which.call_count == 1 + assert mock_which.call_args.args == ("claude",) + assert mock_open.call_count == 2 + assert mock_open.call_args_list[1].args[0][0] == "/usr/local/bin/claude" + + anyio.run(_test) + + def test_guard_is_a_no_op_off_windows(self): + with patch(self._PLATFORM, return_value="Linux"): + SubprocessCLITransport._reject_windows_batch_cli("/odd/claude.cmd") + + def _not_found_message(self, system: str) -> str: + from claude_agent_sdk._errors import CLINotFoundError + + transport = SubprocessCLITransport(prompt="test", options=ClaudeAgentOptions()) + with ( + patch(self._PLATFORM, return_value=system), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + return_value=None, + ), + patch("pathlib.Path.exists", return_value=False), + pytest.raises(CLINotFoundError) as exc_info, + ): + transport._find_cli() + return str(exc_info.value) + + def test_not_found_message_on_windows_recommends_native_exe(self): + # The npm route yields a claude.cmd shim that connect() refuses, so + # the Windows message must lead with the native claude.exe install. + message = self._not_found_message("Windows") + assert "install.ps1" in message + assert "claude.exe" in message + assert message.index("install.ps1") < message.index("npm") + assert "refuses" in message + + def test_not_found_message_off_windows_is_unchanged(self): + message = self._not_found_message("Linux") + assert message.startswith( + "Claude Code not found. Install with:\n" + " npm install -g @anthropic-ai/claude-code\n" + ) + assert "install.ps1" not in message + + def test_fallback_locations_find_native_windows_exe(self): + # The native installer writes ~/.local/bin/claude.exe; Path.exists() + # does no PATHEXT resolution, so the fallback list must probe the + # .exe name explicitly for a stale-PATH process to find it. + from pathlib import Path + + native_exe = Path.home() / ".local/bin/claude.exe" + + def _exists(path: Path) -> bool: + return path == native_exe + + transport = SubprocessCLITransport(prompt="test", options=ClaudeAgentOptions()) + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + return_value=None, + ), + patch("pathlib.Path.exists", new=_exists), + patch("pathlib.Path.is_file", new=_exists), + ): + assert transport._find_cli() == str(native_exe) + + def test_windows_fallback_skips_posix_shaped_probes(self): + # Shim-only Windows machine that also has an extensionless + # ~/.local/bin/claude artifact (WSL / git-bash script) and a + # C:\usr\local\bin\claude planted on the current drive: the Windows + # fallback must probe only the native ~/.local/bin/claude.exe, so + # discovery still hands connect() the shim and the batch-script + # refusal fires (0 spawns) instead of spawning either artifact. + from pathlib import Path + + native_exe = Path.home() / ".local/bin/claude.exe" + + def _exists(path: Path) -> bool: + return path != native_exe + + async def _test(): + from claude_agent_sdk._errors import CLIConnectionError + + shim = "C:\\Users\\u\\AppData\\Roaming\\npm\\claude.CMD" + + def _which(name: str) -> str | None: + return shim if name == "claude" else None + + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + with ( + patch(self._PLATFORM, return_value="Windows"), + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", + side_effect=_which, + ), + patch("pathlib.Path.exists", new=_exists), + patch("pathlib.Path.is_file", new=_exists), + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLIConnectionError, match="batch script"), + ): + await transport.connect() + + assert mock_open.call_count == 0 + + anyio.run(_test) + + +class TestExtraArgsValueBinding: + """extra_args uses the equals form for dash-leading values so the value + binds to its flag instead of parsing as a separate CLI flag.""" + + def test_dash_leading_value_uses_equals_form(self): + transport = SubprocessCLITransport( + prompt="test", + options=make_options(extra_args={"future-flag": "--evil"}), + ) + cmd = transport._build_command() + assert "--future-flag=--evil" in cmd + assert "--evil" not in cmd + assert "--future-flag" not in cmd + + def test_ordinary_value_keeps_two_token_form(self): + transport = SubprocessCLITransport( + prompt="test", + options=make_options( + extra_args={"future-flag": "plain", "bool-flag": None} + ), + ) + cmd = transport._build_command() + idx = cmd.index("--future-flag") + assert cmd[idx + 1] == "plain" + assert "--bool-flag" in cmd + + +class TestWindowsCmdMetacharacterRejection: + """Defense in depth: resume/session_id reject cmd.exe metacharacters on + Windows so those values stay inert even if a cmd.exe hop reappears.""" + + _PLATFORM = "claude_agent_sdk._internal.transport.subprocess_cli.platform.system" + + @pytest.mark.parametrize( + "value", + [ + "x&calc", + "x|whoami", + "xout", + "x^y", + "x%PATH%y", + "x!VAR!y", + 'x"y', + "x\ny", + "x\ry", + ], + ) + def test_bad_resume_values_raise_on_windows(self, value: str): + transport = SubprocessCLITransport( + prompt="test", options=make_options(resume=value) + ) + with ( + patch(self._PLATFORM, return_value="Windows"), + pytest.raises(ValueError, match="unsafe"), + ): + transport._build_command() + + def test_bad_session_id_raises_on_windows(self): + transport = SubprocessCLITransport( + prompt="test", options=make_options(session_id="x&ver") + ) + with ( + patch(self._PLATFORM, return_value="Windows"), + pytest.raises(ValueError, match="session_id"), + ): + transport._build_command() + + def test_ordinary_title_is_accepted_on_windows(self): + title = "My project - daily notes (v2) #3" + transport = SubprocessCLITransport( + prompt="test", options=make_options(resume=title) + ) + with patch(self._PLATFORM, return_value="Windows"): + cmd = transport._build_command() + assert f"--resume={title}" in cmd + + def test_posix_allows_metacharacters(self): + transport = SubprocessCLITransport( + prompt="test", + options=make_options(resume="title & % | notes", session_id="a>b"), + ) + with patch(self._PLATFORM, return_value="Linux"): + cmd = transport._build_command() + assert "--resume=title & % | notes" in cmd + assert "--session-id=a>b" in cmd