From 98f21b5a5984df8063cf4a459c2ead5f0404b12d Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:19:45 +0000 Subject: [PATCH 1/7] feat(internal): add safe executable resolution module Add claude_agent_sdk._internal.executable: find_executable / require_executable / resolve_argv / run and ExecutableNotFoundError. find_executable is a hand-written PATH walk that searches only the fully-absolute entries of PATH -- never the current working directory, neither implicitly (as shutil.which and CreateProcess do on Windows) nor via "."/empty/relative entries -- and on Windows only ever returns a native .exe/.com image, never a .bat/.cmd shim or an extensionless file. A name with a directory part is the caller's explicit choice and is only made absolute. Whatever it returns is absolute and normalised, ready to be handed to the OS as argv[0]. The platform rules are pure ntpath/posixpath string functions, so the Windows behaviour is unit-tested on POSIX CI as well. The module docstring states the invariant and the named guarantees (G1-G5, D1) shared with the other Anthropic SDKs; tests/test_executable.py implements the shared vectors (V1-V12, W1-W3, P1). Motivated by HackerOne #3888880 (CWE-427): the callsites are migrated in the following commit. --- src/claude_agent_sdk/_internal/executable.py | 301 ++++++++++++++ tests/test_executable.py | 415 +++++++++++++++++++ 2 files changed, 716 insertions(+) create mode 100644 src/claude_agent_sdk/_internal/executable.py create mode 100644 tests/test_executable.py diff --git a/src/claude_agent_sdk/_internal/executable.py b/src/claude_agent_sdk/_internal/executable.py new file mode 100644 index 00000000..b7b0940d --- /dev/null +++ b/src/claude_agent_sdk/_internal/executable.py @@ -0,0 +1,301 @@ +r"""Safe executable resolution and invocation for helper programs. + +**Safe executable resolution.** This SDK never launches a helper program by +bare name. Every program it spawns is either an absolute path it constructed +itself, or a bare name resolved by the SDK's own ``find_executable`` -- which +searches only the fully-absolute entries of ``PATH``, never the current +working directory (neither implicitly, as Windows ``CreateProcess``/ +``shutil.which``/libuv do, nor via ``.``/empty/relative ``PATH`` entries), +and on Windows returns only native executables (``.exe``/``.com``). The +absolute path it returns is what is handed to the OS. This holds on every +platform, so a file planted in a directory the user merely *works in* (a +cloned repository, an extracted archive) is never selected as a helper +binary. + +Guarantees (the tests, the docs and the sibling SDKs refer to them by name): + +- G1 -- absolute argv[0]. The program path handed to the OS process-creation + API is always absolute. Bare names never reach ``subprocess`` / ``anyio``. +- G2 -- the current directory is never a search location. Not implicitly + (Windows), not via ``.``, not via an empty ``PATH`` entry (which POSIX + ``execvp`` treats as the current directory), not via a relative entry + (``bin``, ``..\tools``), and on Windows not via a drive-relative + (``C:bin``) or rooted-but-driveless (``\bin``) entry either. Only fully + absolute entries are searched: POSIX ``/...``; Windows ``X:\...`` / + ``X:/...`` or UNC ``\\server\share\...``. One pair of surrounding double + quotes on a Windows entry is stripped first (``"C:\Program Files\Git\cmd"`` + is legal in ``PATH``). ``~`` and ``%VAR%`` / ``$VAR`` are not expanded -- + the OS does not expand them at spawn time either -- so such entries are + skipped as relative. An unset or empty ``PATH`` finds nothing: there is no + ``os.defpath`` fallback. +- G3 -- Windows: native images only. For a bare name the candidates are the + name with each *allowed* extension appended, by default ``.exe`` and + ``.com`` -- never ``.bat`` / ``.cmd`` (``CreateProcess`` runs those through + ``cmd.exe /c``, which re-parses the arguments: the BatBadBut / + CVE-2024-27980 class) and never an extensionless file (WinError 193). A + name that already ends in an allowed extension is tried as-is only; a name + with any other extension has no candidates at all. The allow-list is a + parameter so a caller can *detect* -- not run -- a ``.cmd`` shim in order + to explain why it is refused. On POSIX the name is tried as-is only. +- G4 -- explicit paths are the caller's decision. If the name contains a + path separator (``/``; on Windows also ``\`` or a drive colon) nothing is + searched: the absolute, normalised form is returned iff it is an existing + regular file (and executable, on POSIX), else "not found". ``./tool`` + therefore resolves against the current directory -- the caller asked for + exactly that. +- G5 -- one implementation, enforced. ``shutil.which`` and + ``distutils.spawn.find_executable`` are banned by ruff (the + ``flake8-tidy-imports`` ``banned-api`` table in ``pyproject.toml``), and + ``tests/test_executable_invariant.py`` fails if any process-spawning call + under ``src/`` passes a program name literal that is not an absolute path. +- D1 -- defense in depth, Windows only. The transport module sets + ``NoDefaultCurrentDirectoryInExePath`` in this process's environment at + import, so ``CreateProcess`` (and ``shutil.which`` on Python >= 3.12) stop + searching the current directory for anything downstream of the SDK that + still does an OS-level bare-name lookup. It is not the fix -- older Pythons + ignore it -- G1-G4 are. + +A match is a regular file (symlinks followed, so a directory named like the +tool is skipped) that is executable (``os.access(X_OK)``) on POSIX; on +Windows existence plus an allowed extension is what makes a file runnable. +The result is ``normpath(join(entry, candidate))`` -- deliberately not +``realpath``, which keeps Homebrew / Scoop symlink-farm spellings intact -- +and nothing is cached: ``PATH`` and the current directory change, and the +walk is cheap. + +Why not ``shutil.which(name, path=sanitized)``: on Windows CPython puts the +current directory at the front of the search even when ``path=`` is given +(unconditionally before 3.12; unless ``NoDefaultCurrentDirectoryInExePath`` +is set from 3.12 on) and appends every ``PATHEXT`` extension, ``.bat`` and +``.cmd`` included. So the walk is written out here, with the platform rules +as pure ``ntpath`` / ``posixpath`` string functions that the tests exercise +on every host. + +Mirrors claude-code's ``safeExecutableResolver``; keep in sync with the other +Anthropic SDKs (claude-agent-sdk-python, anthropic-sdk-python, +anthropic-sdk-typescript, anthropic-sdk-go). +""" + +from __future__ import annotations + +import errno +import ntpath +import os +import posixpath +import re +import subprocess +from collections.abc import Iterator, Sequence +from pathlib import Path +from typing import Any, Final + +WINDOWS_NATIVE_EXTENSIONS: Final = (".exe", ".com") + +# G2 on Windows: drive-absolute ("C:\", "c:/") or UNC ("\\server\share"). +_WINDOWS_ABSOLUTE_ENTRY = re.compile(r"[A-Za-z]:[\\/]|[\\/]{2}[^\\/]+[\\/]+[^\\/]+") + + +class ExecutableNotFoundError(FileNotFoundError): + """No executable matched ``name`` (see :func:`require_executable`). + + A :class:`FileNotFoundError`, so callers that already treat a missing + helper program as an ``OSError`` keep working unchanged. + """ + + def __init__(self, name: str) -> None: + super().__init__( + errno.ENOENT, + "executable not found (only absolute PATH entries are searched," + " never the current directory)", + name, + ) + self.name = name + + +def _is_windows() -> bool: + # os.name rather than sys.platform so mypy does not narrow either branch + # away on the host it type-checks on. + return os.name == "nt" + + +def _unquote_path_entry(entry: str, *, windows: bool) -> str: + r"""Strip one pair of surrounding double quotes from a Windows PATH entry. + + ``"C:\Program Files\X"`` is a legal Windows ``PATH`` entry; the quotes + are not part of the directory name. POSIX attaches no meaning to them. + """ + if windows and len(entry) >= 2 and entry[0] == entry[-1] == '"': + return entry[1:-1] + return entry + + +def _is_searchable_path_entry(entry: str, *, windows: bool) -> bool: + r"""G2: whether a ``PATH`` entry is fully absolute, and therefore searched. + + Empty, ``.``, relative, ``~``- or variable-prefixed entries -- and on + Windows drive-relative (``C:bin``) and rooted-but-driveless (``\bin``) + ones -- all resolve against some current directory, so they are skipped. + """ + entry = _unquote_path_entry(entry, windows=windows) + if windows: + return _WINDOWS_ABSOLUTE_ENTRY.match(entry) is not None + return entry.startswith("/") + + +def _candidate_names( + name: str, *, windows: bool, windows_extensions: Sequence[str] +) -> list[str]: + """G3: the file names looked for in each searched directory. + + POSIX: the name as-is. Windows: the name as-is if it already ends in an + allowed extension; nothing at all if it carries any other extension + (``claude.cmd`` and ``tool.js`` are never candidates under the default + allow-list); otherwise the name with each allowed extension appended, in + order -- and never the bare extensionless name. + """ + if not windows: + return [name] + lowered = name.lower() + if any(lowered.endswith(ext.lower()) for ext in windows_extensions): + return [name] + if ntpath.splitext(name)[1]: + return [] + return [name + ext for ext in windows_extensions] + + +def _is_explicit_path(name: str, *, windows: bool) -> bool: + r"""G4: whether ``name`` spells out a location rather than a bare command. + + Any directory part counts, a bare drive (``C:claude``) included. On POSIX + a backslash is an ordinary file-name character, so ``a\b`` is still a + bare name there. + """ + if windows: + return bool(ntpath.dirname(name)) + return bool(posixpath.dirname(name)) + + +def _iter_candidate_paths( + name: str, path: str, *, windows: bool, windows_extensions: Sequence[str] +) -> Iterator[str]: + """Absolute, normalised candidate paths for bare ``name``, in ``PATH`` order. + + G2 and G3 combined, as pure string logic: nothing here touches the + filesystem, so the Windows walk is testable on any host. + """ + names = _candidate_names( + name, windows=windows, windows_extensions=windows_extensions + ) + if not names: + return + for raw_entry in path.split(ntpath.pathsep if windows else posixpath.pathsep): + entry = _unquote_path_entry(raw_entry, windows=windows) + if not _is_searchable_path_entry(entry, windows=windows): + continue + for candidate in names: + if windows: + yield ntpath.normpath(ntpath.join(entry, candidate)) + else: + yield posixpath.normpath(posixpath.join(entry, candidate)) + + +def _is_executable_file(path: str, *, windows: bool) -> bool: + """The match test for a candidate path. + + An existing regular file -- symlinks followed, so a directory named like + the tool never matches -- that is executable on POSIX. On Windows there + is no execute bit to consult: existence plus the allowed extension the + candidate was built with is what makes it runnable. + """ + try: + if not Path(path).is_file(): + return False + except OSError: + # PATH may name places this process cannot even stat (another + # user's directory, a dead network share). That is "not here". + return False + return windows or os.access(path, os.X_OK) + + +def find_executable( + name: str, + *, + path: str | None = None, + windows_extensions: Sequence[str] = WINDOWS_NATIVE_EXTENSIONS, +) -> str | None: + """Resolve ``name`` to the absolute path of an executable, or ``None``. + + A bare name is looked up in the fully-absolute entries of ``path`` + (default: the ``PATH`` environment variable) only -- never in the current + directory, whether implicitly or via ``.`` / empty / relative entries + (G2) -- and on Windows only as a native ``.exe`` / ``.com`` image, or + whatever ``windows_extensions`` allows (G3). A name that contains a path + separator is not searched for: its absolute form is returned iff that + file exists and is executable (G4). Whatever is returned is absolute and + normalised, ready to be handed to the OS as ``argv[0]`` (G1). See the + module docstring for the full contract. + """ + windows = _is_windows() + if name in ("", ".", ".."): + return None + if _is_explicit_path(name, windows=windows): + candidate = ntpath.abspath(name) if windows else posixpath.abspath(name) + return candidate if _is_executable_file(candidate, windows=windows) else None + if path is None: + path = os.environ.get("PATH", "") + for candidate in _iter_candidate_paths( + name, path, windows=windows, windows_extensions=windows_extensions + ): + if _is_executable_file(candidate, windows=windows): + return candidate + return None + + +def require_executable( + name: str, + *, + path: str | None = None, + windows_extensions: Sequence[str] = WINDOWS_NATIVE_EXTENSIONS, +) -> str: + """:func:`find_executable`, raising :class:`ExecutableNotFoundError` on no match.""" + resolved = find_executable(name, path=path, windows_extensions=windows_extensions) + if resolved is None: + raise ExecutableNotFoundError(name) + return resolved + + +def resolve_argv( + argv: Sequence[str | os.PathLike[str]], + *, + path: str | None = None, + windows_extensions: Sequence[str] = WINDOWS_NATIVE_EXTENSIONS, +) -> list[str]: + """``argv`` with ``argv[0]`` replaced by its :func:`require_executable` result. + + The remaining arguments are passed through untouched (G1 concerns the + program only). + """ + if not argv: + raise ValueError("argv must name a program to run") + program, *args = (os.fspath(arg) for arg in argv) + return [ + require_executable(program, path=path, windows_extensions=windows_extensions), + *args, + ] + + +def run( + argv: Sequence[str | os.PathLike[str]], /, **kwargs: Any +) -> subprocess.CompletedProcess[Any]: + """``subprocess.run(resolve_argv(argv), **kwargs)``. + + The one blessed way for SDK code to run a helper program synchronously: + the program is resolved by :func:`require_executable` first, so the OS is + only ever handed an absolute path (G1) and a missing program raises + :class:`ExecutableNotFoundError` before anything is spawned. ``shell`` + and ``executable`` are refused because either would hand program lookup + back to a shell or to the OS. + """ + for unsupported in ("shell", "executable"): + if kwargs.get(unsupported): + raise TypeError(f"run() does not support the {unsupported!r} argument") + return subprocess.run(resolve_argv(argv), **kwargs) diff --git a/tests/test_executable.py b/tests/test_executable.py new file mode 100644 index 00000000..eea3ca7a --- /dev/null +++ b/tests/test_executable.py @@ -0,0 +1,415 @@ +"""Tests for the SDK's safe executable resolution (``_internal/executable.py``). + +The vectors are shared with the other Anthropic SDKs and named after the +guarantees in that module's docstring: V1-V12 exercise the real filesystem on +whatever host runs the suite, W1-W3 exercise the Windows rules as pure string +logic (so they run on POSIX CI too), P1 is the POSIX-only backslash rule. The +per-callsite regression tests live next to their subjects (``_find_cli`` in +test_transport.py, the git worktree lookup in test_sessions.py, the Keychain +read in test_session_resume.py); the G5 enforcement scan is +test_executable_invariant.py. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from claude_agent_sdk._internal import executable +from claude_agent_sdk._internal.executable import ( + WINDOWS_NATIVE_EXTENSIONS, + ExecutableNotFoundError, + find_executable, + require_executable, + resolve_argv, + run, +) + +WINDOWS = os.name == "nt" +posix_only = pytest.mark.skipif(WINDOWS, reason="POSIX execute bit / file names") + +# What the bare name "tool" resolves to on this host: G3 appends .exe on +# Windows and never returns the extensionless file there. +TOOL = "tool.exe" if WINDOWS else "tool" + + +def _make_executable(path: Path, body: str = "#!/bin/sh\nexit 0\n") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body) + path.chmod(0o755) + return path + + +@pytest.fixture +def layout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """``bin/`` holds the real tool. ``plant/`` becomes the current directory + and holds a planted one, plus ``rel/sub/tool`` beneath it. Every location + carries both spellings (``tool`` and ``tool.exe``) so the same vectors run + on every host.""" + for directory in ("bin", "plant", "plant/rel/sub"): + for name in ("tool", "tool.exe"): + _make_executable(tmp_path / directory / name) + monkeypatch.chdir(tmp_path / "plant") + return tmp_path + + +def _join(*entries: str) -> str: + return os.pathsep.join(entries) + + +class TestSearchNeverUsesTheCurrentDirectory: + """G2 on the host platform, against real files (V1-V5, V12).""" + + def test_v1_absolute_entry_is_searched(self, layout: Path) -> None: + found = find_executable("tool", path=str(layout / "bin")) + assert found == str(layout / "bin" / TOOL) + + def test_v2_empty_entry_is_not_the_current_directory(self, layout: Path) -> None: + # POSIX execvp treats an empty PATH entry as "."; we skip it. + found = find_executable("tool", path=_join("", str(layout / "bin"))) + assert found == str(layout / "bin" / TOOL) + + def test_v3_dot_entry_is_skipped(self, layout: Path) -> None: + found = find_executable("tool", path=_join(".", str(layout / "bin"))) + assert found == str(layout / "bin" / TOOL) + + def test_v4_relative_entry_is_skipped(self, layout: Path) -> None: + assert (Path("rel/sub") / TOOL).is_file() # reachable from the cwd... + found = find_executable( + "tool", path=_join(str(Path("rel", "sub")), str(layout / "bin")) + ) + assert found == str(layout / "bin" / TOOL) # ...and still not used + + def test_v5_only_cwd_reachable_entries_find_nothing(self, layout: Path) -> None: + assert Path(TOOL).is_file() # the plant is right here in the cwd + path = _join(".", "", "rel/sub", str(Path("rel", "sub")), "~", "plant") + assert find_executable("tool", path=path) is None + + def test_v12_unset_or_empty_path_finds_nothing( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # No os.defpath fallback and, above all, no fallback to the cwd. + assert find_executable("tool", path="") is None + monkeypatch.delenv("PATH", raising=False) + assert find_executable("tool") is None + + def test_path_defaults_to_the_environment( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("PATH", _join(".", str(layout / "bin"))) + assert find_executable("tool") == str(layout / "bin" / TOOL) + + def test_first_searchable_entry_wins(self, layout: Path) -> None: + _make_executable(layout / "other" / TOOL) + found = find_executable( + "tool", path=_join(str(layout / "other"), str(layout / "bin")) + ) + assert found == str(layout / "other" / TOOL) + + +class TestMatchTest: + """What counts as a hit (V6, V7).""" + + @posix_only + def test_v6_non_executable_file_is_not_a_match(self, layout: Path) -> None: + (layout / "bin" / "tool").chmod(0o644) + assert find_executable("tool", path=str(layout / "bin")) is None + + def test_v7_directory_named_like_the_tool_is_skipped(self, layout: Path) -> None: + (layout / "dirs" / TOOL).mkdir(parents=True) + found = find_executable( + "tool", path=_join(str(layout / "dirs"), str(layout / "bin")) + ) + assert found == str(layout / "bin" / TOOL) + + @posix_only + def test_symlinked_executable_keeps_its_path_spelling(self, layout: Path) -> None: + # Homebrew / Scoop style symlink farms: follow the link for the match + # test, but return the PATH spelling rather than the realpath. + farm = layout / "farm" + farm.mkdir() + (farm / "tool").symlink_to(layout / "bin" / "tool") + assert find_executable("tool", path=str(farm)) == str(farm / "tool") + + def test_unreadable_entry_is_not_an_error( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + real_is_file = Path.is_file + + def flaky_is_file(self: Path) -> bool: + if self.parent.name == "locked": + raise PermissionError("simulated EACCES") + return real_is_file(self) + + monkeypatch.setattr(Path, "is_file", flaky_is_file) + found = find_executable( + "tool", path=_join(str(layout / "locked"), str(layout / "bin")) + ) + assert found == str(layout / "bin" / TOOL) + + +class TestExplicitPaths: + """G4: a name with a directory part is the caller's decision (V8, V9, V11).""" + + def test_v8_dot_slash_resolves_against_the_current_directory( + self, layout: Path + ) -> None: + found = find_executable("./tool", path=str(layout / "bin")) + assert found == str(layout / "plant" / "tool") + + def test_v9_absolute_path_is_returned_as_is(self, layout: Path) -> None: + target = str(layout / "bin" / TOOL) + assert find_executable(target, path="") == target + assert find_executable(str(layout / "bin" / "nope"), path="") is None + + def test_explicit_path_to_a_directory_is_not_a_match(self, layout: Path) -> None: + assert find_executable(str(layout / "bin"), path="") is None + + @posix_only + def test_explicit_non_executable_file_is_not_a_match(self, layout: Path) -> None: + (layout / "plant" / "tool").chmod(0o644) + assert find_executable("./tool", path="") is None + + def test_v11_degenerate_names_find_nothing(self, layout: Path) -> None: + for name in ("", ".", ".."): + assert find_executable(name, path=str(layout / "bin")) is None + + +class TestResultShape: + """G1: whatever comes back can go straight to the OS (V10).""" + + def test_v10_result_is_absolute_and_normalised(self, layout: Path) -> None: + messy = str(layout / "bin") + os.sep + "." + os.sep + found = find_executable("tool", path=messy) + assert found is not None + assert Path(found).is_absolute() + assert found == os.path.normpath(found) + assert found == str(layout / "bin" / TOOL) + + +class TestWindowsRules: + """W1-W3: the Windows walk as pure string logic, runnable on any host.""" + + @pytest.mark.parametrize( + "entry", + [ + "C:\\bin", + "c:/bin", + "C:\\", + "\\\\srv\\share\\bin", + "//srv/share", + '"C:\\Program Files\\X"', + ], + ) + def test_w1_fully_absolute_entries_are_searchable(self, entry: str) -> None: + assert executable._is_searchable_path_entry(entry, windows=True) + + @pytest.mark.parametrize( + "entry", + [ + "", + ".", + "..", + "bin", + "..\\x", + "\\bin", + "/bin", + "C:bin", + "C:", + "~\\bin", + "%SystemRoot%\\system32", + '""', + '"bin"', + ], + ) + def test_w1_everything_else_is_skipped(self, entry: str) -> None: + assert not executable._is_searchable_path_entry(entry, windows=True) + + @pytest.mark.parametrize( + ("name", "extensions", "expected"), + [ + ("rg", WINDOWS_NATIVE_EXTENSIONS, ["rg.exe", "rg.com"]), + ("rg.exe", WINDOWS_NATIVE_EXTENSIONS, ["rg.exe"]), + ("RG.EXE", WINDOWS_NATIVE_EXTENSIONS, ["RG.EXE"]), + ("tool.com", WINDOWS_NATIVE_EXTENSIONS, ["tool.com"]), + ("claude.cmd", WINDOWS_NATIVE_EXTENSIONS, []), + ("claude.bat", WINDOWS_NATIVE_EXTENSIONS, []), + ("tool.js", WINDOWS_NATIVE_EXTENSIONS, []), + ("claude.cmd", (".exe", ".com", ".cmd", ".bat"), ["claude.cmd"]), + ("claude", (".cmd", ".bat"), ["claude.cmd", "claude.bat"]), + ], + ) + def test_w2_candidate_names( + self, name: str, extensions: tuple[str, ...], expected: list[str] + ) -> None: + got = executable._candidate_names( + name, windows=True, windows_extensions=extensions + ) + assert got == expected + + def test_posix_candidate_is_the_name_as_is(self) -> None: + for name in ("rg", "rg.exe", "claude.cmd", "tool.js"): + assert executable._candidate_names( + name, windows=False, windows_extensions=WINDOWS_NATIVE_EXTENSIONS + ) == [name] + + def test_candidate_paths_come_only_from_absolute_entries(self) -> None: + path = ";".join( + ["", ".", "plant", "C:plant", "\\plant", '"C:\\Program Files\\T"', "D:/t"] + ) + got = list( + executable._iter_candidate_paths( + "tool", + path, + windows=True, + windows_extensions=WINDOWS_NATIVE_EXTENSIONS, + ) + ) + assert got == [ + "C:\\Program Files\\T\\tool.exe", + "C:\\Program Files\\T\\tool.com", + "D:\\t\\tool.exe", + "D:\\t\\tool.com", + ] + + @pytest.fixture + def fake_windows(self, monkeypatch: pytest.MonkeyPatch) -> set[str]: + """Run find_executable under the Windows rules against a fake + filesystem: the returned set is what "exists".""" + files: set[str] = set() + monkeypatch.setattr(executable, "_is_windows", lambda: True) + monkeypatch.setattr( + executable, + "_is_executable_file", + lambda path, *, windows: windows and path in files, + ) + return files + + def test_w3_planted_exe_in_cwd_is_never_found(self, fake_windows: set[str]) -> None: + # The cwd is C:\plant. shutil.which would return C:\plant\tool.exe + # for every one of these PATH values; find_executable never looks. + fake_windows.update({"C:\\plant\\tool.exe", "C:\\bin\\tool.exe"}) + assert find_executable("tool", path=".;C:\\bin") == "C:\\bin\\tool.exe" + assert find_executable("tool", path="C:\\bin;.") == "C:\\bin\\tool.exe" + for cwd_only in ("", ".", ";;", ".;plant;C:plant;\\plant"): + assert find_executable("tool", path=cwd_only) is None + + def test_w3_shim_in_earlier_directory_does_not_shadow_native_exe( + self, fake_windows: set[str] + ) -> None: + # npm's directory (claude.cmd) and a git-bash wrapper (extensionless + # claude) precede the native installer's directory on PATH. + # shutil.which walks directory-major with PATHEXT and returns the + # shim; G3 only ever considers claude.exe / claude.com. + fake_windows.update( + { + "C:\\npm\\claude.cmd", + "C:\\npm\\claude", + "C:\\gitbash\\claude", + "C:\\native\\claude.exe", + } + ) + path = 'C:\\npm;C:\\gitbash;"C:\\native"' + assert find_executable("claude", path=path) == "C:\\native\\claude.exe" + assert find_executable("claude.exe", path=path) == "C:\\native\\claude.exe" + # The allow-list parameter lets a caller *detect* the shim (to + # explain why it is refused) without it ever being a spawn candidate. + shim = find_executable("claude", path=path, windows_extensions=(".cmd", ".bat")) + assert shim == "C:\\npm\\claude.cmd" + fake_windows.discard("C:\\native\\claude.exe") + assert find_executable("claude", path=path) is None + + def test_explicit_path_detection(self) -> None: + for name in ("C:\\x\\claude.exe", "x\\claude", "./claude", "C:claude"): + assert executable._is_explicit_path(name, windows=True) + for name in ("claude", "claude.exe"): + assert not executable._is_explicit_path(name, windows=True) + + +class TestPosixRules: + def test_p1_backslash_is_not_a_separator(self) -> None: + assert not executable._is_explicit_path("a\\b", windows=False) + assert executable._is_explicit_path("a/b", windows=False) + + @posix_only + def test_p1_backslash_name_is_searched_as_a_bare_name(self, layout: Path) -> None: + _make_executable(layout / "bin" / "a\\b") + assert find_executable("a\\b", path=str(layout / "bin")) == str( + layout / "bin" / "a\\b" + ) + + @pytest.mark.parametrize("entry", ["/usr/bin", "/", "//double"]) + def test_absolute_entries_are_searchable(self, entry: str) -> None: + assert executable._is_searchable_path_entry(entry, windows=False) + + @pytest.mark.parametrize( + "entry", ["", ".", "..", "bin", "./bin", "~/bin", "$HOME/bin", '"/quoted"'] + ) + def test_everything_else_is_skipped(self, entry: str) -> None: + assert not executable._is_searchable_path_entry(entry, windows=False) + + +class TestRequireResolveRun: + """The entry points built on find_executable.""" + + def test_require_executable_raises_a_file_not_found_error( + self, layout: Path + ) -> None: + assert require_executable("tool", path=str(layout / "bin")) == str( + layout / "bin" / TOOL + ) + with pytest.raises(ExecutableNotFoundError) as exc_info: + require_executable("tool", path=".") + assert isinstance(exc_info.value, FileNotFoundError) + assert exc_info.value.name == "tool" + assert exc_info.value.filename == "tool" + assert "never the current directory" in str(exc_info.value) + + def test_resolve_argv_replaces_only_the_program(self, layout: Path) -> None: + argv = resolve_argv( + ["tool", "--flag", Path("rel/sub")], path=str(layout / "bin") + ) + assert argv == [ + str(layout / "bin" / TOOL), + "--flag", + str(Path("rel", "sub")), + ] + with pytest.raises(ValueError): + resolve_argv([], path=str(layout / "bin")) + + def test_run_resolves_then_spawns(self, layout: Path) -> None: + # sys.executable is absolute, so this is the G4 pass-through. + result = run( + [sys.executable, "-c", "print('spawned')"], capture_output=True, text=True + ) + assert isinstance(result, subprocess.CompletedProcess) + assert result.stdout.strip() == "spawned" + + @posix_only + def test_run_searches_path_for_a_bare_name( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _make_executable(layout / "bin" / "tool", '#!/bin/sh\necho "from-bin:$1"\n') + _make_executable(layout / "plant" / "tool", '#!/bin/sh\necho "PLANTED"\n') + monkeypatch.setenv("PATH", _join(".", "", str(layout / "bin"))) + result = run(["tool", "x"], capture_output=True, text=True) + assert result.stdout.strip() == "from-bin:x" + assert result.args[0] == str(layout / "bin" / "tool") + + def test_run_never_spawns_a_plant( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("PATH", _join(".", "", "rel/sub")) + assert Path(TOOL).is_file() + with pytest.raises(ExecutableNotFoundError): + run(["tool"], capture_output=True) + + def test_run_refuses_shell_and_executable_overrides(self, layout: Path) -> None: + with pytest.raises(TypeError): + run([sys.executable, "-c", "pass"], shell=True) + with pytest.raises(TypeError): + run(["python", "-c", "pass"], executable=sys.executable) From bb88473bd2544b6b22e58073d46ae4fbe706813c Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:37:17 +0000 Subject: [PATCH 2/7] fix: never resolve the CLI or helper programs from the working directory HackerOne #3888880 (CWE-427): on Windows installs without the bundled CLI, _find_cli() returned shutil.which("claude"), and CPython's shutil.which searches the current directory first on win32 -- so a claude.exe planted in whatever directory the application runs from (a cloned repository, an extracted archive) was spawned, twice, with the developer's environment. The same class applied to the bare "git" the session listing runs and, in principle, to a bare options.cli_path. - transport: _find_cli() resolves "claude" with find_executable -- absolute PATH entries only, never the current directory, and on Windows only a native claude.exe/claude.com. That subsumes the old which("claude") / which("claude.exe") / _is_windows_native_exe preference dance (a .cmd shim or extensionless wrapper in an earlier PATH directory can no longer shadow a native exe in a later one, since neither is ever a candidate). A shim-only Windows machine still gets the explanatory batch-script refusal: the shim is detected with an explicit (".cmd", ".bat") allow-list and handed to _reject_windows_batch_cli, never spawned. - transport: connect() settles one absolute CLI path up front (_settle_cli_path) for both the version probe and the main spawn. An explicit cli_path with a directory part is only made absolute (the caller's decision, checked against the batch-script rule as spelled); a bare cli_path="claude" now goes through find_executable instead of being handed to the OS to search. - sessions: `git worktree list` runs through executable.run(), which resolves git safely and raises an OSError subclass when there is none, so the existing "git unavailable -> []" path is unchanged. - session_resume: the macOS Keychain read invokes /usr/bin/security by absolute path (a SIP-protected fixed location) instead of "security". Tests: the Windows discovery tests now stub find_executable instead of shutil.which (the two that exercised CPython's PATHEXT quirks -- "claude.exe.cmd" and the extensionless which() hit -- are replaced by find_executable's own W2/W3 vectors); new regression tests plant claude / claude.exe / claude.cmd / git in the working directory with ".", "" and relative entries on PATH and assert against the real filesystem that nothing planted is ever discovered or run, that a bare cli_path is looked up safely, and that a relative cli_path is settled to a single absolute path. --- .../_internal/session_resume.py | 8 +- src/claude_agent_sdk/_internal/sessions.py | 8 +- .../_internal/transport/subprocess_cli.py | 117 +++--- tests/test_integration.py | 5 +- tests/test_session_resume.py | 23 ++ tests/test_sessions.py | 49 +++ tests/test_transport.py | 332 +++++++++++------- 7 files changed, 366 insertions(+), 176 deletions(-) diff --git a/src/claude_agent_sdk/_internal/session_resume.py b/src/claude_agent_sdk/_internal/session_resume.py index ed23164b..978709ea 100644 --- a/src/claude_agent_sdk/_internal/session_resume.py +++ b/src/claude_agent_sdk/_internal/session_resume.py @@ -47,6 +47,12 @@ # CLAUDE_CONFIG_DIR is unset (production OAUTH_FILE_SUFFIX is empty). _KEYCHAIN_SERVICE_NAME = "Claude Code-credentials" +# The macOS ``security`` tool, by absolute path on purpose (G1 in +# _internal/executable.py): it is a SIP-protected system binary at this fixed +# location on every supported macOS, so nothing is looked up on PATH -- or in +# the working directory -- to read the Keychain. +_SECURITY_BIN = "/usr/bin/security" + @dataclass class MaterializedResume: @@ -413,7 +419,7 @@ def _read_keychain_credentials() -> str | None: try: result = subprocess.run( [ - "security", + _SECURITY_BIN, "find-generic-password", "-a", user, diff --git a/src/claude_agent_sdk/_internal/sessions.py b/src/claude_agent_sdk/_internal/sessions.py index cb1cb134..ee3a424c 100644 --- a/src/claude_agent_sdk/_internal/sessions.py +++ b/src/claude_agent_sdk/_internal/sessions.py @@ -21,6 +21,7 @@ import anyio from ..types import SDKSessionInfo, SessionKey, SessionMessage, SessionStore +from . import executable from .session_store_validation import _store_implements logger = logging.getLogger(__name__) @@ -391,7 +392,12 @@ def _get_worktree_paths(cwd: str) -> list[str]: Returns empty list if git is unavailable or cwd is not in a repo. """ try: - result = subprocess.run( + # executable.run resolves "git" from the absolute PATH entries only + # -- never a git / git.exe sitting in ``cwd`` or in the process's + # own working directory -- and raises ExecutableNotFoundError (an + # OSError) when there is none, so "git unavailable" still lands in + # the except clause below. + result = executable.run( ["git", "worktree", "list", "--porcelain"], cwd=cwd, capture_output=True, diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 1bddfd4d..708fdcaf 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -6,7 +6,6 @@ import os import platform import re -import shutil import signal from collections.abc import AsyncIterable, AsyncIterator from contextlib import suppress @@ -28,6 +27,7 @@ SystemPromptPreset, ) from .._task_compat import TaskHandle, spawn_detached +from ..executable import find_executable from . import Transport logger = logging.getLogger(__name__) @@ -245,34 +245,33 @@ def __init__( self._write_lock: anyio.Lock = anyio.Lock() def _find_cli(self) -> str: - """Find Claude Code CLI binary.""" + """Find Claude Code CLI binary. + + Every path this returns is absolute: the bundled binary sits next + to this package, PATH hits come from find_executable (absolute PATH + entries only, never the current directory -- see + _internal/executable.py), and the fallback locations are anchored + at the home directory or the filesystem root. + """ # First, check for bundled CLI bundled_cli = self._find_bundled_cli() if bundled_cli: return bundled_cli - # Fall back to system-wide search - which_hit: str | None = None - if cli := shutil.which("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 + # Fall back to a PATH search -- find_executable, not shutil.which: + # only absolute PATH entries are searched and never the current + # directory (shutil.which puts it first on Windows, so a claude.exe + # planted in whatever directory the application happens to run from + # used to win), and on Windows only a native claude.exe / claude.com + # is ever returned. That also settles what shutil.which needed a + # preference dance for: it walks PATH directory-major with PATHEXT, + # so npm's claude.cmd shim (which connect() refuses to spawn) or an + # extensionless git-bash / WSL wrapper script (which fails at spawn + # with WinError 193) in an early PATH directory shadowed a native + # claude.exe installed in a later one. Neither is a candidate now, + # so the native executable is found wherever it sits on PATH. + if cli := find_executable("claude"): + return cli if platform.system() == "Windows": # Only the native installer's claude.exe. Path.exists() does @@ -300,14 +299,19 @@ def _find_cli(self) -> str: 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": + # No native executable was discoverable anywhere. Look for npm's + # claude.cmd shim purely to *detect* it -- the extension + # allow-list makes it findable here, nothing runs it -- and + # return it so connect() raises the batch-script refusal (with + # its remediation) for a shim-only machine rather than a bare + # not-found error. An extensionless wrapper script is never + # returned any more: that machine now gets the not-found error + # below, with the native install instructions, instead of an + # opaque WinError 193 at spawn. + shim = find_executable("claude", windows_extensions=(".cmd", ".bat")) + if shim is not None: + return shim # npm's Windows install is a claude.cmd shim, which connect() # refuses (_reject_windows_batch_cli), so do not recommend it. raise CLINotFoundError( @@ -345,15 +349,45 @@ def _find_bundled_cli(self) -> str | None: return None + def _settle_cli_path(self) -> str: + """Return the absolute CLI path connect() will spawn. + + Nothing downstream hands the OS a name to look up (G1 in + _internal/executable.py): discovery (_find_cli) only ever yields + absolute paths, and an explicit options.cli_path is settled here. + A value with a directory part is the caller's decision (G4): it is + made absolute against the current working directory -- once, up + front, so the version probe and the main spawn are guaranteed to + run the same file -- and nothing is searched; a file that is not + there still surfaces as CLINotFoundError from connect(). A bare + command name ("claude") used to be handed to the OS to look up, + and Windows looks in the current directory first; it now goes + through find_executable like discovery does. + """ + if self._cli_path is None: + return self._find_cli() + cli_path = self._cli_path + # Classify the caller's own spelling before it is made absolute: + # _is_windows_batch_cli deliberately judges every component as + # written rather than re-deriving an effective final one (see + # there). connect() vets the settled path again. + self._reject_windows_batch_cli(cli_path) + if self._is_explicit_cli_path(cli_path): + return str(Path(cli_path).absolute()) + if found := find_executable(cli_path): + return found + raise CLINotFoundError(f"Claude Code not found on PATH: {cli_path}") + @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(). + def _is_explicit_cli_path(cli_path: str) -> bool: + """Whether an options.cli_path value spells out a location (the + caller's decision, G4) rather than a bare command name to look up + on PATH. Plain string logic keyed off platform.system(), like + _is_windows_batch_cli, so the Windows rule -- either slash, or a + drive colon -- is exercised on POSIX CI too. """ - name = cli_path.replace("\\", "/").rsplit("/", 1)[-1] - return name.rstrip(". ").lower().endswith((".exe", ".com")) + separators = "/\\:" if platform.system() == "Windows" else "/" + return any(sep in cli_path for sep in separators) @staticmethod def _is_windows_batch_cli(cli_path: str) -> bool: @@ -771,8 +805,9 @@ async def connect(self) -> None: if self._process: return - if self._cli_path is None: - self._cli_path = await anyio.to_thread.run_sync(self._find_cli) + # Absolute from here on: both spawns below (version probe and main + # process) hand the OS this exact path, never a name to search for. + self._cli_path = await anyio.to_thread.run_sync(self._settle_cli_path) # Validate the resolved CLI before anything is spawned with it -- # this guards the version probe below as well as the main spawn. @@ -832,6 +867,7 @@ async def connect(self) -> None: # Pipe stderr only when the caller registered a callback. stderr_dest = PIPE if self._options.stderr is not None else None + # cmd[0] is the absolute path settled in connect() (G1). self._process = await anyio.open_process( cmd, stdin=PIPE, @@ -1130,6 +1166,7 @@ async def _check_claude_version(self) -> None: version_process = None try: with anyio.fail_after(2): # 2 second timeout + # self._cli_path is the absolute path settled in connect() (G1). version_process = await anyio.open_process( [self._cli_path, "-v"], stdout=PIPE, diff --git a/tests/test_integration.py b/tests/test_integration.py index 5b434546..816b042c 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -166,7 +166,10 @@ def test_cli_not_found(self): async def _test(): with ( - patch("shutil.which", return_value=None), + patch( + "claude_agent_sdk._internal.transport.subprocess_cli.find_executable", + return_value=None, + ), patch("pathlib.Path.exists", return_value=False), pytest.raises(CLINotFoundError) as exc_info, ): diff --git a/tests/test_session_resume.py b/tests/test_session_resume.py index e1098e67..ed18bb2b 100644 --- a/tests/test_session_resume.py +++ b/tests/test_session_resume.py @@ -1333,3 +1333,26 @@ async def noop() -> None: cleanup=noop, ) assert m.config_dir == Path("/tmp/x") + + +# --------------------------------------------------------------------------- +# Keychain read uses the system `security` binary by absolute path +# --------------------------------------------------------------------------- + + +def test_keychain_read_runs_usr_bin_security_by_absolute_path() -> None: + """The macOS Keychain fallback must hand the OS an absolute program path + (G1 in _internal/executable.py), never the bare name ``security`` for it + to look up on PATH / in the working directory.""" + from claude_agent_sdk._internal import session_resume + + completed = Mock(returncode=0, stdout='{"claudeAiOauth": {}}\n') + with ( + patch.object(session_resume.platform, "system", return_value="Darwin"), + patch.object(session_resume.subprocess, "run", return_value=completed) as run, + ): + assert session_resume._read_keychain_credentials() == '{"claudeAiOauth": {}}' + + argv = run.call_args.args[0] + assert argv[0] == "/usr/bin/security" + assert argv[1] == "find-generic-password" diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 1f324f00..4c5d7f5f 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -1872,3 +1872,52 @@ def test_empty_agent_file(self, claude_config_dir: Path, tmp_path: Path): (subagents_dir / "agent-empty.jsonl").write_text("") assert get_subagent_messages(sid, "empty", directory=project_path) == [] + + +class TestWorktreeLookupNeverRunsAPlantedGit: + """``_get_worktree_paths`` shells out to git. The git it runs must come + from an absolute PATH entry -- never a ``git`` / ``git.exe`` sitting in + the project directory being listed or in the process's working + directory (HackerOne #3888880's bug class; G1/G2 in + _internal/executable.py).""" + + _MARKER = "planted-git-was-executed" + + @pytest.fixture + def repo(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + for name in ("git", "git.exe"): + planted = repo / name + planted.write_text(f"#!/bin/sh\ntouch '{repo / self._MARKER}'\n") + planted.chmod(0o755) + monkeypatch.chdir(repo) + # Every way of reaching the plant except an absolute entry. + monkeypatch.setenv("PATH", os.pathsep.join(["", ".", str(Path("..", "repo"))])) + return repo + + def test_planted_git_is_not_run(self, repo: Path) -> None: + from claude_agent_sdk._internal.sessions import _get_worktree_paths + + # No git on any absolute PATH entry: "git unavailable" -> [] as before. + assert _get_worktree_paths(str(repo)) == [] + assert not (repo / self._MARKER).exists() + + @pytest.mark.skipif(os.name == "nt", reason="runs a shebang script as git") + def test_git_from_an_absolute_path_entry_is_used( + self, repo: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + from claude_agent_sdk._internal.sessions import _get_worktree_paths + + real_bin = tmp_path / "bin" + real_bin.mkdir() + fake_git = real_bin / "git" + fake_git.write_text( + "#!/bin/sh\n" + 'printf "worktree /work/main\\nHEAD abc\\n\\nworktree /work/wt\\n"\n' + ) + fake_git.chmod(0o755) + monkeypatch.setenv("PATH", os.pathsep.join([".", "", str(real_bin)])) + + assert _get_worktree_paths(str(repo)) == ["/work/main", "/work/wt"] + assert not (repo / self._MARKER).exists() diff --git a/tests/test_transport.py b/tests/test_transport.py index ce700876..27487fb0 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -4,6 +4,7 @@ import uuid from collections.abc import AsyncIterator from contextlib import nullcontext +from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import anyio @@ -14,6 +15,7 @@ DEFAULT_CLI_PATH = "/usr/bin/claude" _ABSENT = object() # sentinel for "field not sent on the wire" +_FIND_EXECUTABLE = "claude_agent_sdk._internal.transport.subprocess_cli.find_executable" def make_options(**kwargs: object) -> ClaudeAgentOptions: @@ -38,10 +40,7 @@ async def _test(): assert transport._cli_path is None with ( - patch( - "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", - return_value=None, - ), + patch(_FIND_EXECUTABLE, return_value=None), patch("pathlib.Path.exists", return_value=False), pytest.raises(CLINotFoundError) as exc_info, ): @@ -2491,112 +2490,48 @@ class TestWindowsBatchScriptRefusal: _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. + @staticmethod + def _fake_find_executable(native: str | None = None, shim: str | None = None): + """A find_executable stand-in for a Windows PATH: ``native`` is what + the default (.exe/.com) lookup yields, ``shim`` what the .cmd/.bat + detection probe yields.""" + from claude_agent_sdk._internal.executable import WINDOWS_NATIVE_EXTENSIONS + + def _find( + name: str, + *, + windows_extensions: tuple[str, ...] = WINDOWS_NATIVE_EXTENSIONS, + ) -> str | None: + assert name == "claude" + if tuple(windows_extensions) == WINDOWS_NATIVE_EXTENSIONS: + return native + assert set(windows_extensions) == {".cmd", ".bat"} + return shim + + return _find + + def test_npm_cmd_shim_is_detected_and_refused(self): + # Shim-only machine: no native claude.exe is discoverable (not on + # PATH, not in the fallback location); only npm's claude.cmd is on + # PATH. find_executable never returns the shim as something to run, + # so discovery probes for it explicitly and hands it to connect() + # purely so the batch-script refusal -- with its remediation -- + # fires instead of a bare not-found error. Nothing is spawned. 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(_FIND_EXECUTABLE, new=self._fake_find_executable(shim=shim)), 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"), @@ -2607,19 +2542,17 @@ def _which(name: str) -> str | None: 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. + def test_native_exe_wins_over_npm_shim(self): + # Dual-install machine: npm's %APPDATA%\npm (claude.cmd) precedes the + # native installer's %USERPROFILE%\.local\bin (claude.exe) on PATH. + # shutil.which walks PATH directory-major with PATHEXT and used to + # return the shim from the earlier directory; find_executable only + # considers claude.exe / claude.com (see test_executable.py, W3), so + # discovery gets the native executable directly, never consults the + # shim probe, and connect() proceeds instead of refusing. 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) + shim = "C:\\Users\\u\\AppData\\Roaming\\npm\\claude.CMD" transport = SubprocessCLITransport( prompt="test", options=ClaudeAgentOptions() @@ -2632,16 +2565,20 @@ def _which(name: str) -> str | None: SubprocessCLITransport, "_find_bundled_cli", return_value=None ), patch( - "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", - side_effect=_which, - ), + _FIND_EXECUTABLE, + side_effect=self._fake_find_executable(native=native, shim=shim), + ) as mock_find, 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() + # The default (native-only) lookup, and no shim probe after it. + assert [c.args for c in mock_find.call_args_list] == [("claude",)] + assert [c.kwargs for c in mock_find.call_args_list] == [{}] assert mock_open.call_count == 2 + assert mock_open.call_args_list[0].args[0][0] == native assert mock_open.call_args_list[1].args[0][0] == native anyio.run(_test) @@ -2751,19 +2688,18 @@ async def _test(): 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, + _FIND_EXECUTABLE, return_value="/usr/local/bin/claude" + ) as mock_find, 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",) + # POSIX discovery uses the find_executable("claude") result + # directly: the .cmd shim probe is a Windows-only branch. + assert mock_find.call_count == 1 + assert mock_find.call_args.args == ("claude",) + assert mock_find.call_args.kwargs == {} assert mock_open.call_count == 2 assert mock_open.call_args_list[1].args[0][0] == "/usr/local/bin/claude" @@ -2782,10 +2718,7 @@ def _not_found_message(self, system: str) -> str: patch.object( SubprocessCLITransport, "_find_bundled_cli", return_value=None ), - patch( - "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", - return_value=None, - ), + patch(_FIND_EXECUTABLE, return_value=None), patch("pathlib.Path.exists", return_value=False), pytest.raises(CLINotFoundError) as exc_info, ): @@ -2826,10 +2759,7 @@ def _exists(path: Path) -> bool: patch.object( SubprocessCLITransport, "_find_bundled_cli", return_value=None ), - patch( - "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", - return_value=None, - ), + patch(_FIND_EXECUTABLE, return_value=None), patch("pathlib.Path.exists", new=_exists), patch("pathlib.Path.is_file", new=_exists), ): @@ -2854,9 +2784,6 @@ async def _test(): 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() ) @@ -2865,10 +2792,7 @@ def _which(name: str) -> str | None: patch.object( SubprocessCLITransport, "_find_bundled_cli", return_value=None ), - patch( - "claude_agent_sdk._internal.transport.subprocess_cli.shutil.which", - side_effect=_which, - ), + patch(_FIND_EXECUTABLE, new=self._fake_find_executable(shim=shim)), patch("pathlib.Path.exists", new=_exists), patch("pathlib.Path.is_file", new=_exists), patch("anyio.open_process", new_callable=AsyncMock) as mock_open, @@ -2881,6 +2805,148 @@ def _which(name: str) -> str | None: anyio.run(_test) +class TestCLIDiscoveryNeverUsesTheWorkingDirectory: + """Regression tests for HackerOne #3888880 against the real filesystem. + + A ``claude`` / ``claude.exe`` / ``claude.cmd`` planted in the directory + the application runs from (a cloned repository, an extracted archive) + must never be what the transport spawns -- not through discovery, not + through ``"."`` / empty / relative PATH entries, and not through a bare + ``cli_path="claude"``. find_executable is NOT mocked here; see + test_executable.py for its own vectors and _internal/executable.py for + the guarantees (G1-G4) these tests lean on. + """ + + _CLI = "claude.exe" if os.name == "nt" else "claude" + _MARKER = "planted-cli-was-executed" + + @pytest.fixture + def plant(self, tmp_path, monkeypatch): + """chdir into a directory full of planted CLIs; PATH reaches it every + way except through an absolute entry.""" + plant = tmp_path / "plant" + (plant / "sub").mkdir(parents=True) + for name in ("claude", "claude.exe", "claude.cmd", "sub/claude"): + planted = plant / name + # If anything ever runs the POSIX plant, it leaves a marker. + planted.write_text(f"#!/bin/sh\ntouch '{plant / self._MARKER}'\n") + planted.chmod(0o755) + monkeypatch.chdir(plant) + monkeypatch.setenv("PATH", os.pathsep.join(["", ".", "sub"])) + return plant + + @pytest.fixture + def real_bin(self, tmp_path): + real_bin = tmp_path / "bin" + real_bin.mkdir() + for name in ("claude", "claude.exe"): + (real_bin / name).write_text("#!/bin/sh\nexit 0\n") + (real_bin / name).chmod(0o755) + return real_bin + + @staticmethod + def _no_bundled_or_fallback_cli(): + # Path.exists -> False hides the bundled binary and the well-known + # fallback locations (which may exist on a developer machine); + # find_executable itself does not go through Path.exists. + return ( + patch.object( + SubprocessCLITransport, "_find_bundled_cli", return_value=None + ), + patch("pathlib.Path.exists", return_value=False), + ) + + def test_planted_cli_is_not_discovered(self, plant, real_bin, monkeypatch): + from claude_agent_sdk._errors import CLINotFoundError + + transport = SubprocessCLITransport(prompt="test", options=ClaudeAgentOptions()) + no_bundled, no_fallback = self._no_bundled_or_fallback_cli() + with no_bundled, no_fallback: + with pytest.raises(CLINotFoundError): + transport._find_cli() + + # A real install later on PATH is found straight past the plant. + monkeypatch.setenv("PATH", os.pathsep.join(["", ".", str(real_bin)])) + assert transport._find_cli() == str(real_bin / self._CLI) + + def test_connect_never_spawns_the_plant(self, plant): + from claude_agent_sdk._errors import CLINotFoundError + + async def _test(): + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions() + ) + no_bundled, no_fallback = self._no_bundled_or_fallback_cli() + with ( + no_bundled, + no_fallback, + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLINotFoundError), + ): + await transport.connect() + + assert mock_open.call_count == 0 + assert not (plant / self._MARKER).exists() + + anyio.run(_test) + + def test_bare_cli_path_is_looked_up_on_path_not_in_cwd( + self, plant, real_bin, monkeypatch + ): + # options.cli_path="claude" used to reach the OS as a bare name, and + # Windows resolves that against the current directory first. + from claude_agent_sdk._errors import CLINotFoundError + + async def _refused(): + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions(cli_path="claude") + ) + with ( + patch("anyio.open_process", new_callable=AsyncMock) as mock_open, + pytest.raises(CLINotFoundError, match="not found on PATH"), + ): + await transport.connect() + assert mock_open.call_count == 0 + + anyio.run(_refused) + + monkeypatch.setenv("PATH", os.pathsep.join([".", str(real_bin)])) + + async def _resolved(): + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions(cli_path="claude") + ) + version_process, main_process = _mock_connect_processes() + with patch("anyio.open_process", new_callable=AsyncMock) as mock_open: + mock_open.side_effect = [version_process, main_process] + await transport.connect() + spawned = [c.args[0][0] for c in mock_open.call_args_list] + assert spawned == [str(real_bin / self._CLI)] * 2 + + anyio.run(_resolved) + assert not (plant / self._MARKER).exists() + + def test_relative_cli_path_is_settled_to_one_absolute_path(self, plant): + # An explicit relative path is the caller's decision (G4) and keeps + # pointing where it pointed -- but it is made absolute once, up + # front, so the version probe and the main spawn run the same file + # and the OS is never handed anything to resolve. + async def _test(): + transport = SubprocessCLITransport( + prompt="test", options=ClaudeAgentOptions(cli_path="./my-claude") + ) + version_process, main_process = _mock_connect_processes() + with patch("anyio.open_process", new_callable=AsyncMock) as mock_open: + mock_open.side_effect = [version_process, main_process] + await transport.connect() + + spawned = [c.args[0][0] for c in mock_open.call_args_list] + assert spawned == [str(plant / "my-claude")] * 2 + assert Path(spawned[0]).is_absolute() + + 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.""" From 62f1408fa29b654864037d51dee75f2bd86864fa Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:43:15 +0000 Subject: [PATCH 3/7] build: enforce safe executable resolution (ruff banned-api + AST test) G5 -- one implementation, enforced: - pyproject.toml: enable ruff TID251 and ban shutil.which and distutils.spawn.find_executable with a message pointing at claude_agent_sdk._internal.executable. Only scripts/ (build tooling that runs before the package is importable) is exempt. - tests/test_executable_invariant.py: parses every module under src/ and fails if a subprocess / anyio / asyncio / os.exec* / os.spawn* / os.system-style call names its program with a bare or relative string literal (or a module-level string constant), or if an ambient lookup API is imported or referenced. Parametrized bad/good snippets prove the scanner is not vacuous; run against origin/main it flags exactly the four sites the previous commit migrated. --- pyproject.toml | 17 +++ tests/test_executable_invariant.py | 228 +++++++++++++++++++++++++++++ 2 files changed, 245 insertions(+) create mode 100644 tests/test_executable_invariant.py diff --git a/pyproject.toml b/pyproject.toml index 72359df8..bc444709 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -127,10 +127,27 @@ select = [ "C4", # flake8-comprehensions "PTH", # flake8-use-pathlib "SIM", # flake8-simplify + "TID251", # flake8-tidy-imports banned-api (see the table below) ] ignore = [ "E501", # line too long (handled by formatter) ] +[tool.ruff.lint.flake8-tidy-imports.banned-api] +# Safe executable resolution (G5 in src/claude_agent_sdk/_internal/executable.py): +# ambient program lookup is banned. Both of these search the current working +# directory on Windows (CWE-427, HackerOne #3888880) and neither sticks to +# native .exe/.com images. tests/test_executable_invariant.py additionally +# fails on any process-spawning call under src/ that names its program by a +# bare or relative literal. +"shutil.which".msg = "shutil.which searches the current directory on Windows (CWE-427). Use claude_agent_sdk._internal.executable.find_executable / require_executable / run instead." +"distutils.spawn.find_executable".msg = "Searches the current directory on Windows (CWE-427). Use claude_agent_sdk._internal.executable.find_executable instead." + +[tool.ruff.lint.per-file-ignores] +# Build tooling runs before the package and its dependencies are importable, +# and only on CI runners / maintainer machines, so it cannot (and need not) +# go through claude_agent_sdk._internal.executable. +"scripts/**" = ["TID251"] + [tool.ruff.lint.isort] known-first-party = ["claude_agent_sdk"] \ No newline at end of file diff --git a/tests/test_executable_invariant.py b/tests/test_executable_invariant.py new file mode 100644 index 00000000..8d217ce6 --- /dev/null +++ b/tests/test_executable_invariant.py @@ -0,0 +1,228 @@ +"""G5 -- the safe-executable invariant is enforced, not just documented. + +Every module under ``src/`` is parsed and the suite fails if + +* a process-spawning call (``subprocess.*``, ``anyio.open_process`` / + ``run_process``, ``asyncio.create_subprocess_*``, ``os.system`` / + ``popen`` / ``exec*`` / ``spawn*`` / ``posix_spawn*``) names its program + with a string literal -- or a module-level ``NAME = "literal"`` constant -- + that is not an absolute path, i.e. hands the OS a bare name to go and + search for (G1); or +* an ambient-search API (``shutil.which``, + ``distutils.spawn.find_executable``) is imported or referenced at all. + +The blessed route is ``claude_agent_sdk._internal.executable`` +(``find_executable`` / ``require_executable`` / ``run``); see that module's +docstring for the guarantees. ruff's ``banned-api`` table in pyproject.toml +reports the second bullet at lint time already; this test covers both at +test time, and the parametrized cases at the bottom prove the scanner is not +vacuous. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path + +import pytest + +SRC = Path(__file__).parent.parent / "src" + +# Process-spawning calls -> index of the positional argument that names the +# program to run (or the argv / command line that starts with it). +_VARIANTS = ("l", "le", "lp", "lpe", "v", "ve", "vp", "vpe") +_SPAWN_APIS: dict[str, int] = { + "subprocess.run": 0, + "subprocess.Popen": 0, + "subprocess.call": 0, + "subprocess.check_call": 0, + "subprocess.check_output": 0, + "subprocess.getoutput": 0, + "subprocess.getstatusoutput": 0, + "anyio.open_process": 0, + "anyio.run_process": 0, + "asyncio.create_subprocess_exec": 0, + "asyncio.create_subprocess_shell": 0, + "os.system": 0, + "os.popen": 0, + "os.startfile": 0, + "os.posix_spawn": 0, + "os.posix_spawnp": 0, + **{f"os.exec{variant}": 0 for variant in _VARIANTS}, + # os.spawn*(mode, file, ...) + **{f"os.spawn{variant}": 1 for variant in _VARIANTS}, +} +_PROGRAM_KEYWORDS = frozenset({"args", "program", "command", "cmd", "file", "path"}) + +# Never to be used for locating programs: both search the current directory +# on Windows (and neither honours the .exe/.com-only rule). +_BANNED_LOOKUPS = frozenset({"shutil.which", "distutils.spawn.find_executable"}) + +_ABSOLUTE = re.compile(r"/|[A-Za-z]:[\\/]|[\\/]{2}[^\\/]") + +_HINT = "resolve it via claude_agent_sdk._internal.executable (find_executable / run)" + + +def _imports(tree: ast.AST) -> dict[str, str]: + """Local name -> dotted target for the file's absolute imports.""" + names: dict[str, str] = {} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.asname: + names[alias.asname] = alias.name + else: + top = alias.name.split(".")[0] + names[top] = top + elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0: + for alias in node.names: + names[alias.asname or alias.name] = f"{node.module}.{alias.name}" + return names + + +def _string_constants(tree: ast.Module) -> dict[str, str]: + """Module-level ``NAME = "literal"`` (or annotated) assignments.""" + constants: dict[str, str] = {} + for node in tree.body: + if isinstance(node, ast.Assign): + targets, value = node.targets, node.value + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets, value = [node.target], node.value + else: + continue + if isinstance(value, ast.Constant) and isinstance(value.value, str): + for target in targets: + if isinstance(target, ast.Name): + constants[target.id] = value.value + return constants + + +def _dotted(node: ast.AST, imports: dict[str, str]) -> str | None: + if isinstance(node, ast.Name): + return imports.get(node.id, node.id) + if isinstance(node, ast.Attribute): + base = _dotted(node.value, imports) + return f"{base}.{node.attr}" if base is not None else None + return None + + +def _program_literal(arg: ast.expr, constants: dict[str, str]) -> str | None: + """The program a spawn call names, when it is statically knowable.""" + if isinstance(arg, (ast.List, ast.Tuple)): + if not arg.elts or isinstance(arg.elts[0], ast.Starred): + return None + arg = arg.elts[0] + if isinstance(arg, ast.Name): + return constants.get(arg.id) + if isinstance(arg, ast.Constant) and isinstance(arg.value, str): + return arg.value + return None + + +def violations(source: str, filename: str = "") -> list[str]: + """Every breach of the invariant in one module's source.""" + tree = ast.parse(source, filename) + imports = _imports(tree) + constants = _string_constants(tree) + found: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + for alias in node.names: + if f"{node.module}.{alias.name}" in _BANNED_LOOKUPS: + found.append( + f"{filename}:{node.lineno}: imports {node.module}.{alias.name}" + f" -- {_HINT}" + ) + elif ( + isinstance(node, ast.Attribute) + and _dotted(node, imports) in _BANNED_LOOKUPS + ): + found.append( + f"{filename}:{node.lineno}: uses {_dotted(node, imports)} -- {_HINT}" + ) + elif isinstance(node, ast.Call): + api = _dotted(node.func, imports) + if api is None or api not in _SPAWN_APIS: + continue + index = _SPAWN_APIS[api] + if len(node.args) > index: + first: ast.expr | None = node.args[index] + else: + first = next( + (kw.value for kw in node.keywords if kw.arg in _PROGRAM_KEYWORDS), + None, + ) + if first is None: + continue + program = _program_literal(first, constants) + if program is not None and _ABSOLUTE.match(program) is None: + found.append( + f"{filename}:{node.lineno}: {api}() launches {program!r} by" + f" bare/relative name -- {_HINT}" + ) + return found + + +def test_the_sdk_source_upholds_the_invariant() -> None: + modules = sorted(SRC.rglob("*.py")) + assert len(modules) > 10, f"did not find the package under {SRC}" + problems = [ + problem + for module in modules + for problem in violations( + module.read_text(encoding="utf-8"), str(module.relative_to(SRC.parent)) + ) + ] + assert not problems, "Safe executable resolution (G1/G5) violated:\n" + "\n".join( + problems + ) + + +@pytest.mark.parametrize( + "snippet", + [ + 'import subprocess\nsubprocess.run(["git", "status"])', + 'import subprocess as sp\nsp.Popen(("claude", "-v"))', + 'from subprocess import check_output\ncheck_output(["security", "find-generic-password"])', + 'import subprocess\nsubprocess.run(args=["git", "worktree", "list"])', + 'import subprocess\nGIT = "git"\nsubprocess.run([GIT, "status"], check=False)', + 'import subprocess\nsubprocess.run(["./claude"])', + 'import subprocess\nsubprocess.run(["bin\\\\tool.exe"])', + 'import subprocess\nsubprocess.check_call("make")', + 'import anyio\nasync def f():\n await anyio.open_process(["claude", "-v"])', + 'from anyio import run_process\nasync def f():\n await run_process(["rg", "x"])', + 'import asyncio\nasync def f():\n await asyncio.create_subprocess_exec("rg", "--json")', + 'import os\nos.execvp("git", ["git", "status"])', + 'import os\nos.spawnlp(os.P_WAIT, "git", "git")', + 'import os\nos.system("git status")', + 'import os\nos.popen("tar tzf x.tgz")', + 'import shutil\nshutil.which("claude")', + 'import shutil as sh\ncli = sh.which("claude")', + "from shutil import which", + "from distutils.spawn import find_executable", + ], +) +def test_scanner_flags_bare_program_names_and_ambient_lookups(snippet: str) -> None: + assert violations(snippet), snippet + + +@pytest.mark.parametrize( + "snippet", + [ + 'import subprocess\nsubprocess.run(["/usr/bin/security", "find-generic-password"])', + 'import subprocess\n_BIN = "/usr/bin/security"\nsubprocess.run([_BIN, "-h"])', + 'import subprocess\nsubprocess.run(["C:\\\\Windows\\\\System32\\\\where.exe", "git"])', + 'import subprocess\nsubprocess.run([r"\\\\server\\share\\tool.exe"])', + 'import subprocess, sys\nsubprocess.run([sys.executable, "-c", "pass"])', + "import anyio\nasync def f(cmd):\n await anyio.open_process(cmd)", + "import subprocess\ndef f(cli_path):\n return subprocess.run([cli_path, '-v'])", + 'from claude_agent_sdk._internal import executable\nexecutable.run(["git", "status"])', + 'from claude_agent_sdk._internal.executable import find_executable\nfind_executable("claude")', + "import shutil\nshutil.copyfile('a', 'b')", + ], +) +def test_scanner_accepts_absolute_programs_and_the_blessed_helpers( + snippet: str, +) -> None: + assert not violations(snippet), violations(snippet) From 78c1545ca4fd9101896ea540875ff1d73a84d0ee Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:48:45 +0000 Subject: [PATCH 4/7] fix(transport): set NoDefaultCurrentDirectoryInExePath on Windows at import D1 -- defense in depth, deliberately in its own commit; it is NOT the fix for HackerOne #3888880 (the previous commits are: the SDK no longer asks the OS to search for any program). Importing the transport module on Windows now does os.environ.setdefault("NoDefaultCurrentDirectoryInExePath", "1"), mirroring the TypeScript Agent SDK, which sets the same variable process-wide at import. With it present, CreateProcess -- and shutil.which on Python >= 3.12 -- stop searching the current directory for bare command names, which also covers the CLI's own child processes since they inherit it. Python < 3.12's shutil.which ignores it, which is why it cannot be the fix. A value the application set itself is left alone. --- .../_internal/transport/subprocess_cli.py | 25 +++++++++++ tests/test_transport.py | 41 +++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 708fdcaf..d99f6ecf 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -59,6 +59,31 @@ def _kill_active_children() -> None: atexit.register(_kill_active_children) +# Its mere presence in the environment makes CreateProcess (and shutil.which +# on Python >= 3.12) stop searching the current directory for bare command +# names; Windows ignores the value. +_NO_CWD_IN_EXE_SEARCH = "NoDefaultCurrentDirectoryInExePath" + + +def _disable_cwd_executable_search() -> None: + """D1 -- defense in depth, Windows only; NOT the fix. + + The fix is that this SDK never asks the OS to search for a program at + all (G1-G4 in _internal/executable.py). This additionally sets + NoDefaultCurrentDirectoryInExePath process-wide, mirroring the + TypeScript Agent SDK, which does the same at import, so that anything + downstream of the SDK that still performs an OS-level bare-name lookup + -- the CLI's own children included, since they inherit it -- skips the + current directory too. Older Pythons' shutil.which ignores it, which is + exactly why it cannot be the fix. setdefault: a value the application + chose itself is left alone. + """ + if platform.system() == "Windows": + os.environ.setdefault(_NO_CWD_IN_EXE_SEARCH, "1") + + +_disable_cwd_executable_search() + # Parentheses and commas are delimiters to the --allowedTools tokenizer; # control characters (C0, DEL, C1) never appear in a skill directory name. diff --git a/tests/test_transport.py b/tests/test_transport.py index 27487fb0..15e5d05f 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -2462,6 +2462,47 @@ async def _test() -> None: anyio.run(_test) +class TestNoDefaultCurrentDirectoryInExePath: + """D1: importing the transport sets NoDefaultCurrentDirectoryInExePath on + Windows -- defense in depth for anything downstream that still lets the + OS search for a bare command name. Not the fix (see + TestCLIDiscoveryNeverUsesTheWorkingDirectory for that).""" + + _PLATFORM = "claude_agent_sdk._internal.transport.subprocess_cli.platform.system" + _VAR = "NoDefaultCurrentDirectoryInExePath" + + def test_set_on_windows(self, monkeypatch): + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.delenv(self._VAR, raising=False) + with patch(self._PLATFORM, return_value="Windows"): + subprocess_cli._disable_cwd_executable_search() + assert os.environ[self._VAR] == "1" + + def test_existing_value_is_left_alone(self, monkeypatch): + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.setenv(self._VAR, "already-set") + with patch(self._PLATFORM, return_value="Windows"): + subprocess_cli._disable_cwd_executable_search() + assert os.environ[self._VAR] == "already-set" + + def test_not_touched_off_windows(self, monkeypatch): + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.delenv(self._VAR, raising=False) + with patch(self._PLATFORM, return_value="Linux"): + subprocess_cli._disable_cwd_executable_search() + assert self._VAR not in os.environ + + @pytest.mark.skipif(os.name != "nt", reason="checks the real import side effect") + def test_import_side_effect_on_a_real_windows_host(self): + import claude_agent_sdk + + assert claude_agent_sdk.__version__ # importing the package is the point + assert self._VAR in os.environ + + def _mock_connect_processes() -> tuple[MagicMock, MagicMock]: """Build the (version probe, main process) mocks connect() awaits.""" version_process = MagicMock() From 9c896da0201954b3bbe6a042922f9be1837d6472 Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:50:18 +0000 Subject: [PATCH 5/7] docs: document the "never spawn by bare name" rule for contributors Add a "Spawning External Programs" section to the README's Development section and to CLAUDE.md: use _internal/executable.py (find_executable / require_executable / run) instead of bare program names or shutil.which, why (CWE-427: the working directory is searched on Windows), and how CI enforces it (ruff TID251 banned-api + tests/test_executable_invariant.py). CHANGELOG.md is intentionally untouched: this repo generates release entries at publish time (docs: update changelog for vX.Y.Z), and tests/test_changelog.py only admits "## X.Y.Z" headings. --- CLAUDE.md | 5 +++++ README.md | 9 +++++++++ 2 files changed, 14 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index bcf91290..30f05b6d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,3 +25,8 @@ python -m pytest tests/test_client.py - `_internal/` - Internal implementation details - `transport/subprocess_cli.py` - CLI subprocess management - `message_parser.py` - Message parsing logic + - `executable.py` - Safe executable resolution (`find_executable` / `run`); the only way SDK code may locate or spawn a helper program + +# Spawning External Programs + +Never pass a bare program name (`"git"`, `"claude"`) to `subprocess`, `anyio.open_process` or any other process API, and never use `shutil.which`: on Windows both search the current working directory, so a binary planted in the directory the application runs from would be executed (CWE-427). Resolve the program with `claude_agent_sdk._internal.executable.find_executable` / `require_executable`, or run it with `executable.run([...])`, and hand the OS the absolute path. See that module's docstring for the guarantees (G1-G5, D1). CI enforces this: ruff bans `shutil.which` (`TID251` via the `banned-api` table in `pyproject.toml`) and `tests/test_executable_invariant.py` fails on any spawn call under `src/` that names its program by a bare or relative literal. diff --git a/README.md b/README.md index 8d844523..d6c37918 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,15 @@ If you're contributing to this project, run the initial setup script to install This installs a pre-push hook that runs lint checks before pushing, matching the CI workflow. To skip the hook temporarily, use `git push --no-verify`. +### Spawning External Programs + +SDK code never launches a helper program by bare name. Do not pass `"git"`, `"claude"` or any other bare command name to `subprocess`, `anyio.open_process` or another process API, and do not use `shutil.which`: on Windows both search the current working directory, so a binary planted in whatever directory the application happens to run from (a cloned repository, an extracted archive) would be executed instead of the installed one (CWE-427). Instead: + +- resolve the program with `find_executable()` / `require_executable()` from [`src/claude_agent_sdk/_internal/executable.py`](src/claude_agent_sdk/_internal/executable.py), which search only the absolute entries of `PATH` (never the current directory) and on Windows return only native `.exe`/`.com` images, then spawn the absolute path they return; or +- run it synchronously with `executable.run([...], **subprocess_kwargs)`, which does both. + +An absolute path the SDK constructs itself (the bundled CLI, `/usr/bin/security`) is fine as-is. The module docstring spells out the guarantees (G1-G5, D1), which are shared with the other Anthropic SDKs. CI enforces the rule: ruff bans `shutil.which` (rule `TID251`, configured in the `banned-api` table in `pyproject.toml`), and `tests/test_executable_invariant.py` fails if any process-spawning call under `src/` names its program with a bare or relative literal. + ### Building Wheels Locally To build wheels with the bundled Claude Code CLI: From c3374ab4db5dc3ebe3ce31a14c9c6c6f79a3d334 Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 21:53:02 +0000 Subject: [PATCH 6/7] fix(internal): keep ExecutableNotFoundError picklable OSError reconstructs as cls(errno, strerror, filename) when unpickled; give the one-argument subclass a matching __reduce__ so it survives multiprocessing / pytest-xdist boundaries. --- src/claude_agent_sdk/_internal/executable.py | 5 +++++ tests/test_executable.py | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/src/claude_agent_sdk/_internal/executable.py b/src/claude_agent_sdk/_internal/executable.py index b7b0940d..edcac96a 100644 --- a/src/claude_agent_sdk/_internal/executable.py +++ b/src/claude_agent_sdk/_internal/executable.py @@ -110,6 +110,11 @@ def __init__(self, name: str) -> None: ) self.name = name + def __reduce__(self) -> tuple[type[ExecutableNotFoundError], tuple[str]]: + # OSError pickles as cls(*args) == cls(errno, strerror, filename), + # which this one-argument constructor would reject. + return (type(self), (self.name,)) + def _is_windows() -> bool: # os.name rather than sys.platform so mypy does not narrow either branch diff --git a/tests/test_executable.py b/tests/test_executable.py index eea3ca7a..8a7ea47d 100644 --- a/tests/test_executable.py +++ b/tests/test_executable.py @@ -13,6 +13,7 @@ from __future__ import annotations import os +import pickle import subprocess import sys from pathlib import Path @@ -368,6 +369,10 @@ def test_require_executable_raises_a_file_not_found_error( assert exc_info.value.name == "tool" assert exc_info.value.filename == "tool" assert "never the current directory" in str(exc_info.value) + # Survives pickling (multiprocessing, xdist) despite OSError's + # three-argument reconstruction protocol. + clone = pickle.loads(pickle.dumps(exc_info.value)) + assert isinstance(clone, ExecutableNotFoundError) and clone.name == "tool" def test_resolve_argv_replaces_only_the_program(self, layout: Path) -> None: argv = resolve_argv( From 8ffa35d1c3941f4728c69f1d40000f8cdb265344 Mon Sep 17 00:00:00 2001 From: seanyeoh Date: Wed, 5 Aug 2026 22:32:57 +0000 Subject: [PATCH 7/7] fix(internal): harden run()/resolve_argv() edges for parity with anthropic-sdk-python Behaviour parity with the sibling module (anthropic-sdk-python src/anthropic/lib/_executable.py, "harden the resolver's run/resolve_argv edges found in review"); the modules are meant to stay behaviourally identical across SDKs: - run() resolves argv[0] against the PATH the child will actually see: the one in an explicit env={...} when it carries a PATH, else this process's -- matching what subprocess.run itself searches on POSIX, still absolute entries only, and never falling back to os.defpath. - resolve_argv() (and therefore run()) rejects a plain string argv with a clear TypeError instead of "resolving" its first character. Tests ported with the sibling's names/vectors: test_run_resolves_against_the_path_the_child_will_see, test_executable_not_found_error_survives_pickling (split out; adds the copy.copy and errno/str round-trip checks), and the str-argv assertions in test_resolve_argv_replaces_only_the_program. D1 stays process-wide in this repo (no per-child env helper). --- src/claude_agent_sdk/_internal/executable.py | 28 +++++++++-- tests/test_executable.py | 49 ++++++++++++++++++-- 2 files changed, 70 insertions(+), 7 deletions(-) diff --git a/src/claude_agent_sdk/_internal/executable.py b/src/claude_agent_sdk/_internal/executable.py index edcac96a..665301f4 100644 --- a/src/claude_agent_sdk/_internal/executable.py +++ b/src/claude_agent_sdk/_internal/executable.py @@ -84,7 +84,7 @@ import posixpath import re import subprocess -from collections.abc import Iterator, Sequence +from collections.abc import Iterator, Mapping, Sequence from pathlib import Path from typing import Any, Final @@ -277,8 +277,14 @@ def resolve_argv( """``argv`` with ``argv[0]`` replaced by its :func:`require_executable` result. The remaining arguments are passed through untouched (G1 concerns the - program only). + program only). Use this in front of the async spawn APIs + (``anyio.open_process``), which are deliberately not wrapped here. """ + if isinstance(argv, (str, bytes)): + # A ``str`` is a ``Sequence[str]`` to the type checker; iterating it + # would "resolve" its first character. Shell command lines are not + # supported. + raise TypeError("argv must be a sequence of program arguments, not a string") if not argv: raise ValueError("argv must name a program to run") program, *args = (os.fspath(arg) for arg in argv) @@ -299,8 +305,24 @@ def run( :class:`ExecutableNotFoundError` before anything is spawned. ``shell`` and ``executable`` are refused because either would hand program lookup back to a shell or to the OS. + + ``argv[0]`` is resolved against the ``PATH`` the child will see, as + ``subprocess.run`` itself would search on POSIX: the one in + ``kwargs["env"]`` when the caller passes an environment that carries one, + else this process's. Either way only its absolute entries count (G2), and + an ``env`` without any ``PATH`` falls back to this process's ``PATH`` -- + never to ``os.defpath``, which is what CPython would quietly search. """ for unsupported in ("shell", "executable"): if kwargs.get(unsupported): raise TypeError(f"run() does not support the {unsupported!r} argument") - return subprocess.run(resolve_argv(argv), **kwargs) + resolved = resolve_argv(argv, path=_search_path_in(kwargs.get("env"))) + return subprocess.run(resolved, **kwargs) + + +def _search_path_in(env: object) -> str | None: + """The ``PATH`` inside a caller-supplied child environment, if it carries one.""" + if not isinstance(env, Mapping): + return None + value = env.get("PATH") + return value if isinstance(value, str) else None diff --git a/tests/test_executable.py b/tests/test_executable.py index 8a7ea47d..a7ada258 100644 --- a/tests/test_executable.py +++ b/tests/test_executable.py @@ -12,6 +12,7 @@ from __future__ import annotations +import copy import os import pickle import subprocess @@ -369,10 +370,18 @@ def test_require_executable_raises_a_file_not_found_error( assert exc_info.value.name == "tool" assert exc_info.value.filename == "tool" assert "never the current directory" in str(exc_info.value) - # Survives pickling (multiprocessing, xdist) despite OSError's - # three-argument reconstruction protocol. - clone = pickle.loads(pickle.dumps(exc_info.value)) - assert isinstance(clone, ExecutableNotFoundError) and clone.name == "tool" + + def test_executable_not_found_error_survives_pickling(self) -> None: + """Raised inside a ``ProcessPoolExecutor`` worker it must cross the + process boundary intact (``OSError.__reduce__`` alone would replay + three arguments into the one-argument constructor).""" + err = ExecutableNotFoundError("rg") + # Round-trips an object created right here -- no untrusted pickle data. + for clone in (pickle.loads(pickle.dumps(err)), copy.copy(err)): + assert isinstance(clone, ExecutableNotFoundError) + assert clone.name == "rg" and clone.filename == "rg" + assert clone.errno == err.errno + assert str(clone) == str(err) def test_resolve_argv_replaces_only_the_program(self, layout: Path) -> None: argv = resolve_argv( @@ -385,6 +394,14 @@ def test_resolve_argv_replaces_only_the_program(self, layout: Path) -> None: ] with pytest.raises(ValueError): resolve_argv([], path=str(layout / "bin")) + with pytest.raises(ExecutableNotFoundError): + resolve_argv(["tool"], path=".") + # A shell-style string is a ``Sequence[str]`` to the type checker, but + # never an argv: refuse it rather than "resolve" its first character. + with pytest.raises(TypeError, match="not a string"): + resolve_argv("tool --flag", path=str(layout / "bin")) + with pytest.raises(TypeError, match="not a string"): + run("tool --flag", capture_output=True) def test_run_resolves_then_spawns(self, layout: Path) -> None: # sys.executable is absolute, so this is the G4 pass-through. @@ -405,6 +422,30 @@ def test_run_searches_path_for_a_bare_name( assert result.stdout.strip() == "from-bin:x" assert result.args[0] == str(layout / "bin" / "tool") + @posix_only + def test_run_resolves_against_the_path_the_child_will_see( + self, layout: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Like ``subprocess.run`` on POSIX, an explicit ``env`` carrying a + ``PATH`` is the search path -- still subject to the + absolute-entries-only rule.""" + hello = _make_executable( + layout / "bin" / "hello", '#!/bin/sh\necho "from-bin $0"\n' + ) + monkeypatch.setenv("PATH", "") + child_env = {"PATH": _join(".", str(layout / "bin"))} + result = run(["hello"], env=child_env, capture_output=True, text=True) + assert result.stdout == f"from-bin {hello}\n" + with pytest.raises(ExecutableNotFoundError): + run(["hello"], env={"PATH": "."}, capture_output=True) + # No PATH in the child env: fall back to this process's (empty here) + # -- never os.defpath. + with pytest.raises(ExecutableNotFoundError): + run(["hello"], env={"UNRELATED": "1"}, capture_output=True) + monkeypatch.setenv("PATH", str(layout / "bin")) + result = run(["hello"], env={"UNRELATED": "1"}, capture_output=True, text=True) + assert result.returncode == 0 + def test_run_never_spawns_a_plant( self, layout: Path, monkeypatch: pytest.MonkeyPatch ) -> None: