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: 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/src/claude_agent_sdk/_internal/executable.py b/src/claude_agent_sdk/_internal/executable.py new file mode 100644 index 00000000..665301f4 --- /dev/null +++ b/src/claude_agent_sdk/_internal/executable.py @@ -0,0 +1,328 @@ +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, Mapping, 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 __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 + # 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). 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) + 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. + + ``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") + 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/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..d99f6ecf 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__) @@ -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. @@ -245,34 +270,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 +324,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 +374,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 +830,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 +892,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 +1191,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_executable.py b/tests/test_executable.py new file mode 100644 index 00000000..a7ada258 --- /dev/null +++ b/tests/test_executable.py @@ -0,0 +1,461 @@ +"""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 copy +import os +import pickle +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_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( + ["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")) + 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. + 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") + + @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: + 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) 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) 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..15e5d05f 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, ): @@ -2463,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() @@ -2491,20 +2531,38 @@ 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() ) @@ -2514,89 +2572,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("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 +2583,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 +2606,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 +2729,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 +2759,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 +2800,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 +2825,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 +2833,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 +2846,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."""