diff --git a/scripts/_cli_version_validation.py b/scripts/_cli_version_validation.py new file mode 100644 index 000000000..e276f0e07 --- /dev/null +++ b/scripts/_cli_version_validation.py @@ -0,0 +1,119 @@ +r"""Shared validation for Claude Code CLI version strings. + +update_cli_version.py writes the version into src/claude_agent_sdk/_cli_version.py +and download_cli.py reads it back out and hands it to an installer, so the rule +lives here once rather than in two copies that could drift apart. + +The installer is the authority on what a version may be. install.sh (and +install.ps1) enforce + + ^(stable|latest|[0-9]+\.[0-9]+\.[0-9]+(-[^[:space:]]+)?)$ + +VERSION_PATTERN admits a deliberate *subset* of that: the installer's `-[^\s]+` +suffix would accept quotes, backslashes and semicolons, so the suffix here is +narrowed to the characters real versions use. Never widen it back toward the +installer's -- update_cli_version.py writes this value into a Python string +literal in a real source file. + +"latest" and "stable" are the installer's dist-tags, and they are *moving*: +fine for a download, wrong for a pin, since _cli_version.py is the only record +of which build went into the wheels. Hence ``allow_dist_tag``. + +VERSION_PATTERN is deliberately unanchored, and matched with fullmatch(): with +"^...$" a swap to match() would silently accept a trailing newline ("1.0.0\n"); +unanchored, the same swap accepts an obvious prefix like "1.0.0; id" and fails +loudly in tests. +""" + +import re + +VERSION_PATTERN = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+(?:-[0-9A-Za-z.+-]+)?") + +# Matched exactly: install.sh's grammar is case-sensitive, so "Latest" is not a +# tag it would resolve, and accepting it here would widen what reaches a real +# network install. +DIST_TAGS = ("latest", "stable") + +# Anything word-shaped that is not a version: "next", "beta", "nightly". Named +# so the error can say *why* it was rejected instead of printing a regex. +_DIST_TAG_SHAPED = re.compile(r"[A-Za-z][0-9A-Za-z-]*") + +_SUPPORTED_TAGS = ", ".join(repr(tag) for tag in DIST_TAGS) + + +def _expected(allow_dist_tag: bool) -> str: + """The phrase naming what the caller should have passed instead.""" + if allow_dist_tag: + return f"{_SUPPORTED_TAGS}, or a concrete version" + return "a concrete version" + + +def _rejection(version: str, allow_dist_tag: bool) -> str: + """Why ``version`` is unusable -- the most specific reason that applies. + + The caller prefixes "Invalid : ", so each reason reads as the rest + of that sentence. + """ + candidate = version.strip() + + # Reachable only when a tag is not a legal answer: validate_version() + # returns the tag otherwise. + if candidate in DIST_TAGS: + return ( + f"{candidate!r} is a moving dist-tag, not a concrete version. A pinned " + f"version must name the one build that goes into the wheels. Expected " + f"a version matching {VERSION_PATTERN.pattern}" + ) + + # Only where a tag is a legal answer at all; when pinning, the + # dist-tag-shaped branch below correctly says to use a concrete version. + if allow_dist_tag and candidate.lower() in DIST_TAGS: + return f"{candidate!r}. Did you mean {candidate.lower()!r}?" + + # Never normalized away: the caller asked for something we do not support, + # and silently installing a different string is worse. + if candidate[:1] in ("v", "V") and VERSION_PATTERN.fullmatch(candidate[1:]): + return f"{candidate!r}. Did you mean {candidate[1:]!r}? (no leading 'v')" + + if _DIST_TAG_SHAPED.fullmatch(candidate): + return ( + f"{candidate!r} is not a supported dist-tag; " + f"use {_expected(allow_dist_tag)}" + ) + + # Name the raw value, not the stripped one: if the whitespace is the + # problem, the reader has to be able to see it. + return ( + f"{version!r}. Expected {_expected(allow_dist_tag)} " + f"matching {VERSION_PATTERN.pattern}" + ) + + +def validate_version(version: str, *, source: str, allow_dist_tag: bool) -> str: + """Return the usable form of ``version``, or raise. + + Surrounding whitespace is stripped first -- a trailing "\\n" from a file + read or a "\\r" from a CRLF checkout is unambiguous in intent -- and the + stripped value is what the caller gets back and must use downstream. + + Args: + version: The candidate version string. + source: Where the value came from, named in the error message + (e.g. "CLAUDE_CLI_VERSION"). + allow_dist_tag: Whether a moving dist-tag ("latest", "stable") is + acceptable. It is for a download, which resolves it at install + time; it is not for a value pinned into _cli_version.py. + + Raises: + ValueError: If ``version`` is neither an allowed dist-tag nor a + fullmatch of VERSION_PATTERN. + """ + candidate = version.strip() + + if candidate in DIST_TAGS and allow_dist_tag: + return candidate + + if VERSION_PATTERN.fullmatch(candidate): + return candidate + + raise ValueError(f"Invalid {source}: {_rejection(version, allow_dist_tag)}") diff --git a/scripts/download_cli.py b/scripts/download_cli.py index 7167159b5..c0ac39a75 100755 --- a/scripts/download_cli.py +++ b/scripts/download_cli.py @@ -11,13 +11,55 @@ import shutil import subprocess import sys +import tempfile import time +from collections.abc import Callable from pathlib import Path +from typing import NamedTuple, NoReturn + +# scripts/ is not a package. Running this file directly puts scripts/ on +# sys.path, but loading it by path (as the tests do) does not. Appended, not +# prepended: the tests' entry outlives the import, and a prepended scripts/ +# could shadow a stdlib module for the whole pytest process. +_SCRIPTS_DIR = str(Path(__file__).parent) +if _SCRIPTS_DIR not in sys.path: + sys.path.append(_SCRIPTS_DIR) + +import _cli_version_validation as version_validation # noqa: E402 + +# Re-exported: this module's callers and tests refer to download_cli.VERSION_PATTERN. +VERSION_PATTERN = version_validation.VERSION_PATTERN + +# The Windows installer reads the version and the downloaded script's path out +# of the environment under these names, so neither is part of the command text +# PowerShell parses. `$env:NAME` in argument position expands to exactly one +# argument -- the same argv separation the Unix path gets from +# `["bash", script, version]`. +INSTALL_VERSION_ENV_VAR = "CLAUDE_CLI_INSTALL_VERSION" +INSTALL_SCRIPT_ENV_VAR = "CLAUDE_CLI_INSTALL_SCRIPT" + +# How many times retry_install() runs an attempt before giving up. +MAX_INSTALL_ATTEMPTS = 3 + +# What an unset CLAUDE_CLI_VERSION means, and -- being the installer's own +# default -- the one value both install paths express by passing no argument. +DEFAULT_VERSION = "latest" def get_cli_version() -> str: - """Get the CLI version to download from environment or default.""" - return os.environ.get("CLAUDE_CLI_VERSION", "latest") + """Get the CLI version to download from environment or default. + + Returns the stripped value -- use it, rather than the raw environment + string, for everything downstream. + + Raises: + ValueError: If CLAUDE_CLI_VERSION is set to something other than a + dist-tag ("latest", "stable") or a value matching VERSION_PATTERN. + """ + version = os.environ.get("CLAUDE_CLI_VERSION", DEFAULT_VERSION) + return version_validation.validate_version( + version, source="CLAUDE_CLI_VERSION", allow_dist_tag=True + ) def find_installed_cli() -> Path | None: @@ -50,63 +92,212 @@ def find_installed_cli() -> Path | None: return None -def download_cli() -> None: - """Download Claude Code CLI using the official install script.""" - version = get_cli_version() - system = platform.system() +def run_command(command: list[str], env: dict[str, str] | None = None) -> None: + """Run one install command with no shell and no inherited stdin. + + install.sh runs `claude install`, which branches on `[ -t 0 ]`. Under the + old `curl | bash` its stdin was the pipe, so it never saw a TTY; keep it + that way rather than handing it the caller's terminal. + + ``env`` replaces the child's environment when given; None inherits ours. + """ + subprocess.run( + command, + check=True, + capture_output=True, + stdin=subprocess.DEVNULL, + env=env, + ) + + +def _reject_install_script(script_path: str, reason: str) -> NoReturn: + """Report a downloaded installer we refuse to execute, and exit.""" + head = Path(script_path).read_bytes()[:64] + print( + f"Error: downloaded install script {reason} (first bytes: {head!r}). " + f"Refusing to execute it.", + file=sys.stderr, + ) + sys.exit(1) - print(f"Downloading Claude Code CLI version: {version}") - # Build install command based on platform - if system == "Windows": - # Use PowerShell installer on Windows - if version == "latest": - install_cmd = [ - "powershell", - "-ExecutionPolicy", - "Bypass", - "-Command", - "irm https://claude.ai/install.ps1 | iex", - ] - else: - install_cmd = [ - "powershell", - "-ExecutionPolicy", - "Bypass", - "-Command", - f"& ([scriptblock]::Create((irm https://claude.ai/install.ps1))) {version}", - ] - else: - # --retry-all-errors covers 429 from claude.ai when multiple matrix - # jobs fetch install.sh simultaneously. pipefail propagates curl's - # exit code through the pipe so subprocess.run's check=True catches it. - curl = "curl -fsSL --retry 5 --retry-delay 2 --retry-all-errors https://claude.ai/install.sh" - target = "" if version == "latest" else f" -s {version}" - install_cmd = ["bash", "-c", f"set -o pipefail; {curl} | bash{target}"] +def check_install_script(script_path: str) -> None: + """Reject a downloaded install.sh that is not a shell script. + + claude.ai answers unknown paths with HTTP 200 and an HTML body, which + `curl -f` cannot detect, so check the shebang before executing the file. + A wrong body is deterministic, so this fails fast instead of retrying. + """ + with Path(script_path).open("rb") as f: + magic = f.read(2) + if magic != b"#!": + _reject_install_script(script_path, "does not start with a shebang") + + +def check_powershell_install_script(script_path: str) -> None: + """Reject a downloaded install.ps1 that is not a PowerShell script. + The same HTTP 200 + HTML body is invisible to Invoke-RestMethod. A .ps1 has + no shebang to check, so reject what the error page actually is -- an + XML/HTML document, whose first non-blank character is '<' -- and an empty + body. A wrong body is deterministic, so this fails fast instead of retrying. + + '<' alone would be too blunt: a .ps1 may legitimately open with '<#', the + comment-based-help block real installers start with. No HTML or XML document + begins with that, so exempt it. + """ + body = Path(script_path).read_bytes() + # A UTF-8 BOM is legal in a .ps1 and common in PowerShell-authored files. + stripped = body.removeprefix(b"\xef\xbb\xbf").lstrip() + if not stripped: + _reject_install_script(script_path, "is empty") + if stripped.startswith(b"<") and not stripped.startswith(b"<#"): + _reject_install_script(script_path, "looks like an HTML or XML document") + + +def _decode(stream: bytes | str | None) -> str: + """One captured stream as text. run_command() captures bytes; be tolerant.""" + if stream is None: + return "" + if isinstance(stream, bytes): + return stream.decode(errors="replace") + return stream + + +def retry_install(attempt: Callable[[], None]) -> None: + """Run an install attempt, retrying the whole attempt on command failure.""" # Small jitter to stagger parallel matrix builds hitting the same endpoint time.sleep(random.uniform(0, 5)) - last_err: subprocess.CalledProcessError | None = None - for attempt in range(1, 4): + for attempt_num in range(1, MAX_INSTALL_ATTEMPTS + 1): try: - subprocess.run(install_cmd, check=True, capture_output=True) + attempt() return + except OSError as e: + # The command could not be started at all: no `curl`, no `bash`, no + # `powershell` on PATH. Deterministic -- a second attempt cannot + # make the binary appear -- so fail immediately rather than sleeping + # through three of them and then blaming the download. + print(f"Error: could not run the install command: {e}", file=sys.stderr) + sys.exit(1) except subprocess.CalledProcessError as e: - last_err = e - if attempt < 3: - delay = 2**attempt + if attempt_num == MAX_INSTALL_ATTEMPTS: print( - f"Install attempt {attempt} failed (exit {e.returncode}), " - f"retrying in {delay}s...", + f"Error downloading CLI after {MAX_INSTALL_ATTEMPTS} attempts: {e}", file=sys.stderr, ) - time.sleep(delay) + print(f"stdout: {_decode(e.stdout)}", file=sys.stderr) + print(f"stderr: {_decode(e.stderr)}", file=sys.stderr) + sys.exit(1) + + delay = 2**attempt_num + print( + f"Install attempt {attempt_num} failed (exit {e.returncode}), " + f"retrying in {delay}s...", + file=sys.stderr, + ) + time.sleep(delay) + + +# One command to run: its argv, and the environment to run it with (None +# inherits ours). +_Command = tuple[list[str], dict[str, str] | None] + + +class _InstallPlan(NamedTuple): + """How one platform downloads the installer, checks its body, and runs it.""" + + script_path: str + download: _Command + check: Callable[[str], None] + install: _Command + + +def _powershell(command: str) -> list[str]: + """A PowerShell invocation of ``command``.""" + return ["powershell", "-ExecutionPolicy", "Bypass", "-Command", command] + + +def _windows_plan(tmpdir: str, version: str) -> _InstallPlan: + script_path = str(Path(tmpdir) / "install.ps1") + + download_env = {**os.environ, INSTALL_SCRIPT_ENV_VAR: script_path} + download_cmd = _powershell( + "$ProgressPreference = 'SilentlyContinue'; " + "Invoke-RestMethod -Uri https://claude.ai/install.ps1 " + f"-OutFile $env:{INSTALL_SCRIPT_ENV_VAR}" + ) + + install_env = dict(download_env) + install_command = f"& $env:{INSTALL_SCRIPT_ENV_VAR}" + if version != DEFAULT_VERSION: + install_command += f" $env:{INSTALL_VERSION_ENV_VAR}" + install_env[INSTALL_VERSION_ENV_VAR] = version + + return _InstallPlan( + script_path=script_path, + download=(download_cmd, download_env), + check=check_powershell_install_script, + install=(_powershell(install_command), install_env), + ) + + +def _unix_plan(tmpdir: str, version: str) -> _InstallPlan: + script_path = str(Path(tmpdir) / "install.sh") + + # -L follows the cross-host redirect to the bootstrap script. + # --retry-all-errors covers 429 from claude.ai when multiple matrix jobs + # fetch install.sh simultaneously. + curl_cmd = [ + "curl", + "-fsSL", + "--retry", + "5", + "--retry-delay", + "2", + "--retry-all-errors", + "-o", + script_path, + "https://claude.ai/install.sh", + ] + bash_cmd = ["bash", script_path] + if version != DEFAULT_VERSION: + bash_cmd.append(version) + + return _InstallPlan( + script_path=script_path, + download=(curl_cmd, None), + check=check_install_script, + install=(bash_cmd, None), + ) - print(f"Error downloading CLI after 3 attempts: {last_err}", file=sys.stderr) - print(f"stdout: {last_err.stdout.decode()}", file=sys.stderr) - print(f"stderr: {last_err.stderr.decode()}", file=sys.stderr) - sys.exit(1) + +def download_cli() -> None: + """Download Claude Code CLI using the official install script. + + Both platforms download the installer to a temp file, check its body, and + only then execute it -- rather than piping the response straight into a + shell or into `iex`. The whole sequence is retried, so a truncated body is + never reused across attempts. + + Both pass "latest" by passing no argument at all: it is the installer's own + default. Every other accepted value -- a concrete version, or "stable" -- + goes through the argument path. + """ + version = get_cli_version() + print(f"Downloading Claude Code CLI version: {version}") + + build_plan = _windows_plan if platform.system() == "Windows" else _unix_plan + + with tempfile.TemporaryDirectory() as tmpdir: + plan = build_plan(tmpdir, version) + + def attempt() -> None: + run_command(*plan.download) + plan.check(plan.script_path) + run_command(*plan.install) + + retry_install(attempt) def copy_cli_to_bundle() -> None: @@ -165,4 +356,12 @@ def main() -> None: if __name__ == "__main__": - main() + # This script runs as a build step, so report a bad CLAUDE_CLI_VERSION the + # way every other failure here is reported -- one line on stderr, exit 1 -- + # instead of letting a traceback out. The shared validator keeps raising: + # only the entry point turns it into an exit status. + try: + main() + except ValueError as exc: + print(f"Error: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/scripts/update_cli_version.py b/scripts/update_cli_version.py index 1ef17c7ee..92affedbb 100755 --- a/scripts/update_cli_version.py +++ b/scripts/update_cli_version.py @@ -1,32 +1,71 @@ #!/usr/bin/env python3 """Update Claude Code CLI version in _cli_version.py.""" +import json import re import sys from pathlib import Path +# scripts/ is not a package; see the note in download_cli.py. +_SCRIPTS_DIR = str(Path(__file__).parent) +if _SCRIPTS_DIR not in sys.path: + sys.path.append(_SCRIPTS_DIR) -def update_cli_version(new_version: str) -> None: - """Update CLI version in _cli_version.py.""" - # Update _cli_version.py - version_path = Path("src/claude_agent_sdk/_cli_version.py") +import _cli_version_validation as version_validation # noqa: E402 + +DEFAULT_VERSION_PATH = Path("src/claude_agent_sdk/_cli_version.py") + +# Matches the assignment build_wheel.get_bundled_cli_version() reads back. +ASSIGNMENT_PATTERN = re.compile(r'__cli_version__ = "[^"]+"') + + +def update_cli_version(new_version: str, version_path: Path | None = None) -> None: + """Update CLI version in _cli_version.py. + + Raises: + ValueError: If ``new_version`` is not a concrete version, or if the + target file has no ``__cli_version__`` assignment to replace. The + file is left untouched in both cases. + """ + # Validate before touching the file: this writes into a real source file + # that later gets imported, so an unvalidated value closes the string + # literal and injects arbitrary Python. The validated value is the input + # with surrounding whitespace stripped -- write that, not the raw input. + new_version = version_validation.validate_version( + new_version, source="CLI version", allow_dist_tag=False + ) + + if version_path is None: + version_path = DEFAULT_VERSION_PATH content = version_path.read_text() - content = re.sub( - r'__cli_version__ = "[^"]+"', - f'__cli_version__ = "{new_version}"', - content, - count=1, - flags=re.MULTILINE, + # json.dumps() rather than an f-string: it always emits a closed, + # double-quoted, fully escaped literal, so a widened VERSION_PATTERN could + # never make this file unparseable or inject code. repr() is not an option + # -- it emits single quotes, which build_wheel.py's reader would not match. + literal = f"__cli_version__ = {json.dumps(new_version)}" + + # A callable replacement, because re.sub() applies backslash-escape + # processing to a *string* replacement -- \1 and \g<0> expand, \n becomes a + # newline, and a bare trailing \ raises. A callable's return value is used + # literally. + new_content, count = ASSIGNMENT_PATTERN.subn( + lambda _match: literal, content, count=1 ) + if count != 1: + raise ValueError(f"No __cli_version__ assignment found in {version_path}") - version_path.write_text(content) + version_path.write_text(new_content) print(f"Updated {version_path} to {new_version}") if __name__ == "__main__": if len(sys.argv) != 2: - print("Usage: python scripts/update_cli_version.py ") + print("Usage: python scripts/update_cli_version.py ", file=sys.stderr) sys.exit(1) - update_cli_version(sys.argv[1]) + try: + update_cli_version(sys.argv[1]) + except ValueError as exc: + print(f"Error: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/tests/test_download_cli.py b/tests/test_download_cli.py new file mode 100644 index 000000000..6d2d5f68c --- /dev/null +++ b/tests/test_download_cli.py @@ -0,0 +1,688 @@ +"""Tests for scripts/download_cli.py version validation and install invocation.""" + +import importlib.util +import os +import subprocess +import sys +from collections.abc import Callable, Iterator +from contextlib import AbstractContextManager, contextmanager +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +SCRIPT_PATH = Path(__file__).parent.parent / "scripts" / "download_cli.py" + +# scripts/ is not a package, so load download_cli.py by path +_spec = importlib.util.spec_from_file_location("download_cli", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +download_cli = importlib.util.module_from_spec(_spec) +sys.modules["download_cli"] = download_cli +_spec.loader.exec_module(download_cli) + +DEV_VERSION = "2.1.146-dev.20260519.t105443.shaece3dab" +ENV_VAR = download_cli.INSTALL_VERSION_ENV_VAR +SCRIPT_ENV_VAR = download_cli.INSTALL_SCRIPT_ENV_VAR + + +class TestGetCliVersion: + """CLAUDE_CLI_VERSION must be a dist-tag or a concrete version. + + The grammar mirrors the installer's own + (`stable|latest|N.N.N(-suffix)?`), narrowed: anything this accepts but the + installer rejects is an error deferred to install time, where it surfaces + behind a retry loop as a misleading download failure. + """ + + def test_default_is_latest(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CLAUDE_CLI_VERSION", raising=False) + assert download_cli.get_cli_version() == "latest" + + @pytest.mark.parametrize( + "version", + [ + "latest", + # `stable` is a moving tag the installer resolves, like `latest`: + # allowed here (a download resolves it at install time) and + # rejected by update_cli_version.py (a pin must name one build). + "stable", + "1.2.3", + "2.1.195", + DEV_VERSION, + "1.2.3-beta.1", + "1.2.3-rc.1+build.4", + ], + ) + def test_accepted(self, monkeypatch: pytest.MonkeyPatch, version: str) -> None: + monkeypatch.setenv("CLAUDE_CLI_VERSION", version) + assert download_cli.get_cli_version() == version + + @pytest.mark.parametrize( + ("version", "expected"), + [ + ("1.2.3\n", "1.2.3"), + ("1.2.3\r\n", "1.2.3"), + (" 1.2.3 ", "1.2.3"), + ("\tlatest\n", "latest"), + (f" {DEV_VERSION}\r\n", DEV_VERSION), + ], + ) + def test_surrounding_whitespace_is_stripped( + self, monkeypatch: pytest.MonkeyPatch, version: str, expected: str + ) -> None: + monkeypatch.setenv("CLAUDE_CLI_VERSION", version) + assert download_cli.get_cli_version() == expected + + @pytest.mark.parametrize( + ("version", "expected_message"), + [ + # Injection and flag shapes. + ("1.0.0; touch /tmp/pwned", None), + ("--help", None), + ("-s", None), + ("$(id)", None), + ("`id`", None), + ("1.0.0 && id", None), + ("1.0.0 | id", None), + ("1.0.0\nid", None), + ("1.0.0 2.0.0", None), + ("$VERSION", None), + ("../../etc/passwd", None), + ("", None), + (".1.2.3", None), + # Not versions at all -- the old allowlist took these for concrete + # versions and let the installer reject them 11 seconds later. + ("0", None), + ("1.2", None), + ("1.2.3.4", None), + ("1.2.3+build.4", None), # the installer has no bare "+" suffix + # The likeliest typo is told what to type, never silently rewritten + # into a different version than the caller asked for. + ("v2.1.207", "Did you mean '2.1.207'? (no leading 'v')"), + ("V1.2.3", "Did you mean '1.2.3'? (no leading 'v')"), + # Word-shaped, but not a tag the installer resolves. + ("next", "not a supported dist-tag; use 'latest', 'stable'"), + ("beta", "not a supported dist-tag; use 'latest', 'stable'"), + ("nightly", "not a supported dist-tag; use 'latest', 'stable'"), + # The installer's grammar is case-sensitive, so "Latest" is not a + # tag it would resolve: reject it -- naming the spelling that works + # -- rather than quietly turning it into a live install. + ("LATEST", "Did you mean 'latest'?"), + ("Latest", "Did you mean 'latest'?"), + ("STABLE", "Did you mean 'stable'?"), + ("Stable", "Did you mean 'stable'?"), + ], + ) + def test_rejected( + self, + monkeypatch: pytest.MonkeyPatch, + version: str, + expected_message: str | None, + ) -> None: + monkeypatch.setenv("CLAUDE_CLI_VERSION", version) + with pytest.raises(ValueError, match="Invalid CLAUDE_CLI_VERSION") as excinfo: + download_cli.get_cli_version() + + message = str(excinfo.value) + # The offending value is always named, so the reader can see it. + assert repr(version) in message or version in message + if expected_message is not None: + assert expected_message in message + + @pytest.mark.parametrize("version", ["2.1.207", DEV_VERSION, "1.0.0-alpha"]) + def test_pattern_admits_every_published_version_shape(self, version: str) -> None: + """The tightened pattern is about false *accepts*; it must not start + rejecting anything real, releases or dev builds.""" + assert download_cli.VERSION_PATTERN.fullmatch(version) + + @pytest.mark.parametrize( + "char", + [" ", ";", "$", "`", '"', "'", "(", ")", "&", "|", "\n", "\r"], + ) + def test_version_pattern_admits_no_shell_metacharacters(self, char: str) -> None: + """No install path depends on this any more -- the version is an argv + element on Unix and an environment variable on Windows -- but a version + that can express `;` or `$` is not a version. Keep it as defense in + depth: widening the pattern must not let any of these through.""" + assert not download_cli.VERSION_PATTERN.fullmatch(char) + assert not download_cli.VERSION_PATTERN.fullmatch(f"1.2.3{char}") + assert not download_cli.VERSION_PATTERN.fullmatch(f"1.2.3{char}whoami") + + def test_version_pattern_is_unanchored(self) -> None: + """The pattern must stay unanchored so a future swap of fullmatch() for + match() fails loudly on a prefix, rather than silently reintroducing + the trailing-newline bypass that `^...$` + match() allows.""" + assert "^" not in download_cli.VERSION_PATTERN.pattern + assert "$" not in download_cli.VERSION_PATTERN.pattern + assert not download_cli.VERSION_PATTERN.fullmatch("1.0.0\n") + + +@pytest.fixture +def no_sleep() -> Iterator[None]: + """Skip the jitter and retry sleeps.""" + with patch.object(download_cli.time, "sleep"): + yield + + +SHEBANG_BODY = b"#!/bin/bash\necho install\n" +PS_BODY = b"# install.ps1\nWrite-Host installing\n" + +_RunSideEffect = Callable[..., MagicMock] + + +def _fake_curl(body: bytes = SHEBANG_BODY) -> _RunSideEffect: + """A subprocess.run side effect that makes `curl -o PATH` write ``body``.""" + + def side_effect(command: list[str], **kwargs: object) -> MagicMock: + if command[0] == "curl" and "-o" in command: + Path(command[command.index("-o") + 1]).write_bytes(body) + return MagicMock() + + return side_effect + + +def _fake_irm(body: bytes = PS_BODY) -> _RunSideEffect: + """A subprocess.run side effect making the download step write ``body``. + + The Windows download step names its output file only in the environment, so + that is where the fake finds the path to write -- the same indirection the + real PowerShell command resolves. + """ + + def side_effect(command: list[str], **kwargs: Any) -> MagicMock: + env = kwargs.get("env") or {} + path = env.get(SCRIPT_ENV_VAR) + if path and "Invoke-RestMethod" in command[-1]: + Path(path).write_bytes(body) + return MagicMock() + + return side_effect + + +@contextmanager +def _install( + system: str, version: str, side_effect: _RunSideEffect +) -> Iterator[MagicMock]: + """Pin download_cli() to one platform's path, yielding the patched run. + + The two platform paths differ in what they run, not in how they are driven: + both are reached by fixing platform.system(), stubbing subprocess.run, and + putting the version in the environment. That harness lives here once. + """ + with ( + patch.object(download_cli.platform, "system", return_value=system), + patch.object(download_cli.subprocess, "run", side_effect=side_effect) as run, + patch.dict(download_cli.os.environ, {"CLAUDE_CLI_VERSION": version}), + ): + yield run + + +def _unix_install( + version: str, side_effect: _RunSideEffect | None = None +) -> AbstractContextManager[MagicMock]: + """Pin download_cli() to the Unix path; ``side_effect`` defaults to a curl + that writes a real shebang script.""" + return _install("Linux", version, side_effect or _fake_curl()) + + +def _windows_install( + version: str, side_effect: _RunSideEffect | None = None +) -> AbstractContextManager[MagicMock]: + """Pin download_cli() to the Windows path, yielding the patched run.""" + return _install("Windows", version, side_effect or _fake_irm()) + + +def _run_unix_download(version: str, body: bytes = SHEBANG_BODY) -> list[list[str]]: + """Run download_cli() on the Unix path, returning the argv of each subprocess.""" + with _unix_install(version, _fake_curl(body)) as mock_run: + download_cli.download_cli() + return [call.args[0] for call in mock_run.call_args_list] + + +def _run_windows_download(version: str, body: bytes = PS_BODY) -> tuple[Any, Any]: + """Run download_cli() on Windows, returning its (download, install) calls.""" + with _windows_install(version, _fake_irm(body)) as mock_run: + download_cli.download_cli() + + download_call, install_call = mock_run.call_args_list + return download_call, install_call + + +@pytest.mark.usefixtures("no_sleep") +@pytest.mark.parametrize( + ("installer", "side_effect"), + [(_unix_install, _fake_curl()), (_windows_install, _fake_irm())], + ids=["unix", "windows"], +) +def test_no_command_uses_a_shell_or_inherits_stdin( + installer: Callable[..., AbstractContextManager[MagicMock]], + side_effect: _RunSideEffect, +) -> None: + """install.sh runs `claude install`, which branches on `[ -t 0 ]`. The old + `curl | bash` gave it a pipe; it must never inherit a real TTY. And nothing + may reintroduce a shell, on either platform.""" + with installer("1.2.3", side_effect) as mock_run: + download_cli.download_cli() + + assert mock_run.call_args_list + for call in mock_run.call_args_list: + assert call.kwargs["stdin"] is subprocess.DEVNULL, ( + f"{call.args[0][0]} may inherit the caller's TTY" + ) + assert call.kwargs.get("shell") is not True + assert call.kwargs["check"] is True + + +@pytest.mark.usefixtures("no_sleep") +class TestUnixInstall: + """The Unix path downloads install.sh, then executes it without a shell.""" + + def test_downloads_then_executes_script(self) -> None: + curl_cmd, bash_cmd = _run_unix_download("1.2.3") + + assert curl_cmd[0] == "curl" + assert curl_cmd[-1] == "https://claude.ai/install.sh" + assert bash_cmd[0] == "bash" + + # curl writes install.sh into a temp dir; bash executes that same file. + script_path = curl_cmd[curl_cmd.index("-o") + 1] + assert Path(script_path).name == "install.sh" + assert bash_cmd[1] == script_path + + def test_retry_flags_preserved(self) -> None: + curl_cmd, _ = _run_unix_download("1.2.3") + assert "--retry-all-errors" in curl_cmd + assert curl_cmd[curl_cmd.index("--retry") + 1] == "5" + assert curl_cmd[curl_cmd.index("--retry-delay") + 1] == "2" + + @pytest.mark.parametrize("flag", ["f", "s", "S", "L"]) + def test_curl_short_flags_present(self, flag: str) -> None: + """`install.sh` is a cross-host 302, so -L is required; -f/-s/-S keep the + original error and quiet behavior. Checked per letter rather than as the + literal "-fsSL" so splitting the cluster can't silently drop one.""" + curl_cmd, _ = _run_unix_download("1.2.3") + clusters = [ + arg[1:] + for arg in curl_cmd + if arg.startswith("-") and not arg.startswith("--") + ] + assert any(flag in cluster for cluster in clusters), ( + f"curl -{flag} missing from {curl_cmd!r}" + ) + + def test_latest_passes_no_version_argument(self) -> None: + _, bash_cmd = _run_unix_download("latest") + assert bash_cmd[0] == "bash" + assert len(bash_cmd) == 2 + + def test_stable_is_passed_as_an_argument(self) -> None: + """`latest` is the installer's default and so is expressed by passing + nothing; every other accepted value, `stable` included, is an argv + element the installer resolves.""" + _, bash_cmd = _run_unix_download("stable") + assert bash_cmd == ["bash", bash_cmd[1], "stable"] + + @pytest.mark.parametrize("version", ["1.2.3", DEV_VERSION]) + def test_version_is_its_own_argv_element(self, version: str) -> None: + """Regression guard: the version must never be interpolated into a + command string, and the Unix path must never invoke a shell.""" + commands = _run_unix_download(version) + + for cmd in commands: + assert isinstance(cmd, list), f"argv must be a list, got {cmd!r}" + # No `bash -c ` / `sh -c ` anywhere. + assert "-c" not in cmd, f"shell string reintroduced: {cmd!r}" + # The version may only appear as a standalone argv element. + for arg in cmd: + assert arg == version or version not in arg, ( + f"version interpolated into {arg!r}" + ) + + bash_cmd = commands[-1] + assert bash_cmd == ["bash", bash_cmd[1], version] + + @pytest.mark.parametrize( + "body", + [ + b"\nNot found", + b"", + b"#", + b"\x7fELF", + ], + ) + def test_non_shebang_body_is_rejected_before_bash( + self, body: bytes, capsys: pytest.CaptureFixture[str] + ) -> None: + """claude.ai serves HTTP 200 + HTML for unknown paths, which `curl -f` + cannot detect. Such a body must never reach bash.""" + with ( + _unix_install("1.2.3", _fake_curl(body)) as mock_run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + assert "does not start with a shebang" in capsys.readouterr().err + # curl ran once, bash never; the bad body is not retried. + assert [call.args[0][0] for call in mock_run.call_args_list] == ["curl"] + + def test_shebang_body_is_executed(self) -> None: + commands = _run_unix_download("1.2.3", body=b"#!/bin/sh\nexit 0\n") + assert [cmd[0] for cmd in commands] == ["curl", "bash"] + + def test_curl_failure_is_not_masked(self) -> None: + """A failing download must fail the build, not fall through to bash.""" + error = subprocess.CalledProcessError(1, ["curl"], output=b"", stderr=b"boom") + + def fake_run(command: list[str], **kwargs: object) -> MagicMock: + if command[0] == "curl": + raise error + return MagicMock() + + with ( + _unix_install("1.2.3", fake_run) as run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + # curl attempted 3 times, bash never reached. + assert [call.args[0][0] for call in run.call_args_list] == ["curl"] * 3 + + def test_a_failing_install_retries_the_whole_attempt( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """A failing bash re-runs curl too: the script is re-downloaded, so a + truncated or half-written body is not reused across attempts.""" + error = subprocess.CalledProcessError( + 1, ["bash"], output=b"", stderr=b"curl: (6) could not resolve\n" + ) + + curl = _fake_curl() + + def fake_run(command: list[str], **kwargs: object) -> MagicMock: + if command[0] == "curl": + return curl(command, **kwargs) + raise error + + with ( + _unix_install("1.2.3", fake_run) as run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + assert [call.args[0][0] for call in run.call_args_list] == [ + "curl", + "bash", + ] * download_cli.MAX_INSTALL_ATTEMPTS + err = capsys.readouterr().err + assert f"after {download_cli.MAX_INSTALL_ATTEMPTS} attempts" in err + assert "could not resolve" in err + + +@pytest.mark.usefixtures("no_sleep") +class TestWindowsInstall: + """The PowerShell branch downloads install.ps1, checks the body, then runs + it -- the same download / verify / execute shape as the Unix path -- and + routes the version through the same validation.""" + + def test_rejects_injected_version_before_running_anything(self) -> None: + with ( + patch.object(download_cli.platform, "system", return_value="Windows"), + patch.object(download_cli.subprocess, "run") as mock_run, + patch.dict( + download_cli.os.environ, + {"CLAUDE_CLI_VERSION": "1.0.0; Write-Host pwned"}, + ), + pytest.raises(ValueError, match="Invalid CLAUDE_CLI_VERSION"), + ): + download_cli.download_cli() + + mock_run.assert_not_called() + + def test_downloads_then_executes_script(self) -> None: + download_call, install_call = _run_windows_download("1.2.3") + + download_cmd = download_call.args[0] + assert download_cmd[0] == "powershell" + assert ( + "Invoke-RestMethod -Uri https://claude.ai/install.ps1" in download_cmd[-1] + ) + + # The download writes install.ps1 into a temp dir; the install step runs + # that same file, both reaching it through the environment. + script_path = download_call.kwargs["env"][SCRIPT_ENV_VAR] + assert Path(script_path).name == "install.ps1" + assert install_call.kwargs["env"][SCRIPT_ENV_VAR] == script_path + assert install_call.args[0][-1].startswith(f"& $env:{SCRIPT_ENV_VAR}") + + def test_script_path_is_never_in_the_powershell_command_text(self) -> None: + """A temp path can contain spaces (Windows: "C:\\Users\\Foo Bar\\..."), + so it goes through the environment too, never into the command text.""" + download_call, install_call = _run_windows_download("1.2.3") + + script_path = download_call.kwargs["env"][SCRIPT_ENV_VAR] + for call in (download_call, install_call): + for arg in call.args[0]: + assert script_path not in arg, f"script path interpolated into {arg!r}" + + @pytest.mark.parametrize("version", ["1.2.3", DEV_VERSION]) + def test_version_is_carried_in_the_environment_not_the_command_text( + self, version: str + ) -> None: + """Regression guard: the version must reach PowerShell through the + environment, never spliced into the `-Command` string it parses. The + name in the command text and the name in the environment are the same + constant, so a rename cannot desynchronize them silently.""" + download_call, install_call = _run_windows_download(version) + + for call in (download_call, install_call): + for arg in call.args[0]: + assert version not in arg, f"version interpolated into {arg!r}" + + env = install_call.kwargs["env"] + assert env is not None, "PowerShell must be given an explicit environment" + assert env[ENV_VAR] == version + assert f"$env:{ENV_VAR}" in install_call.args[0][-1] + # The child still needs PATH, SystemRoot, etc. to run at all. + assert "CLAUDE_CLI_VERSION" in env + + def test_latest_passes_no_version_argument(self) -> None: + """`latest` invokes the installer with no argument at all -- not with an + empty string -- and sets no version in the child's environment.""" + _, install_call = _run_windows_download("latest") + + command = install_call.args[0][-1] + assert command == f"& $env:{SCRIPT_ENV_VAR}" + assert ENV_VAR not in command + assert ENV_VAR not in install_call.kwargs["env"] + + def test_stable_is_passed_as_an_argument(self) -> None: + _, install_call = _run_windows_download("stable") + assert install_call.args[0][-1] == (f"& $env:{SCRIPT_ENV_VAR} $env:{ENV_VAR}") + assert install_call.kwargs["env"][ENV_VAR] == "stable" + + def test_never_pipes_the_response_body_into_iex(self) -> None: + """Regression guard for the body-integrity gap: an unchecked body must + never be executed straight off the wire.""" + for version in ("latest", "1.2.3"): + for call in _run_windows_download(version): + command = call.args[0][-1] + assert "iex" not in command + assert "scriptblock" not in command + + @pytest.mark.parametrize( + "body", + [ + b"\nNot found", + b"\n oops", + b'', + b"", + b" \n\t\n", + ], + ) + def test_non_powershell_body_is_rejected_before_execution( + self, body: bytes, capsys: pytest.CaptureFixture[str] + ) -> None: + """claude.ai answers unknown paths with HTTP 200 + HTML, which + Invoke-RestMethod cannot detect either. Such a body must never be run, + and -- being deterministic -- must not be retried.""" + with ( + _windows_install("1.2.3", _fake_irm(body)) as mock_run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + assert "Refusing to execute it" in capsys.readouterr().err + # The download ran once; the installer never, and nothing was retried. + assert len(mock_run.call_args_list) == 1 + + @pytest.mark.parametrize( + "body", + [ + PS_BODY, + # A UTF-8 BOM is legal, and common, in a real .ps1. + b"\xef\xbb\xbf# install.ps1\nWrite-Host hi\n", + # A block comment opens with '<' -- and about_Comment_Based_Help + # puts exactly such a help block at the top of a script, which is + # what real installers do. The '<' check must not mistake it for an + # HTML error page. + b"<#\n.SYNOPSIS\nInstalls Claude Code.\n#>\nparam([string]$Version)\n", + b"\xef\xbb\xbf\r\n<#\n.SYNOPSIS\nInstalls Claude Code.\n#>\n", + ], + ) + def test_powershell_body_is_executed(self, body: bytes) -> None: + """A valid installer body reaches PowerShell.""" + download_call, install_call = _run_windows_download("1.2.3", body=body) + assert "Invoke-RestMethod" in download_call.args[0][-1] + assert install_call.args[0][-1].startswith(f"& $env:{SCRIPT_ENV_VAR}") + + +def _raise(error: Exception) -> _RunSideEffect: + def side_effect(command: list[str], **kwargs: object) -> MagicMock: + raise error + + return side_effect + + +@pytest.mark.usefixtures("no_sleep") +class TestMissingBinaryFailsFast: + """A command that cannot be started at all is deterministic. + + The install commands are exec'd directly, so a missing binary raises + FileNotFoundError rather than exiting 127 through a shell. That must be a + one-line error and an exit 1 -- not a traceback, and not three attempts and + a misleading "Error downloading CLI after 3 attempts". + """ + + @pytest.mark.parametrize( + ("installer", "error"), + [ + (_unix_install, FileNotFoundError(2, "No such file or directory", "curl")), + ( + _windows_install, + FileNotFoundError(2, "No such file or directory", "powershell"), + ), + # A noexec tmpdir, say, is just as deterministic as a missing binary. + (_unix_install, PermissionError(13, "Permission denied", "curl")), + ], + ids=["missing-curl", "missing-powershell", "permission-denied"], + ) + def test_first_command_failing_to_start_is_not_retried( + self, + installer: Callable[..., AbstractContextManager[MagicMock]], + error: Exception, + capsys: pytest.CaptureFixture[str], + ) -> None: + with ( + installer("1.2.3", _raise(error)) as mock_run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + assert len(mock_run.call_args_list) == 1 + err = capsys.readouterr().err + assert "could not run the install command" in err + assert "after 3 attempts" not in err + + def test_missing_bash_is_not_retried( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """The installer itself failing to start, after a successful download.""" + curl = _fake_curl() + + def side_effect(command: list[str], **kwargs: object) -> MagicMock: + if command[0] == "curl": + return curl(command, **kwargs) + raise FileNotFoundError(2, "No such file or directory", "bash") + + with ( + _unix_install("1.2.3", side_effect) as mock_run, + pytest.raises(SystemExit) as excinfo, + ): + download_cli.download_cli() + + assert excinfo.value.code == 1 + assert [call.args[0][0] for call in mock_run.call_args_list] == ["curl", "bash"] + assert "could not run the install command" in capsys.readouterr().err + + +class TestCommandLine: + """The script itself reports a bad version rather than raising through. + + build_wheel.py runs this file as `python scripts/download_cli.py`, so an + uncaught ValueError would surface as a traceback in a build log, and the + shared validation module has to import with scripts/ not a package. + + Only rejected versions are exercised: they fail in get_cli_version() before + any curl or installer runs, so no test here touches the network. PATH is + emptied as a second belt -- if validation ever regressed, the run would hit + a missing `curl` rather than really installing the CLI. + """ + + def _run(self, tmp_path: Path, version: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(SCRIPT_PATH)], + env={**os.environ, "CLAUDE_CLI_VERSION": version, "PATH": str(tmp_path)}, + capture_output=True, + text=True, + timeout=60, + ) + + def test_invalid_version_exits_nonzero_without_a_traceback( + self, tmp_path: Path + ) -> None: + result = self._run(tmp_path, "1.0.0; id") + + assert result.returncode == 1 + assert "Invalid CLAUDE_CLI_VERSION" in result.stderr + assert "Traceback" not in result.stderr + assert "ModuleNotFoundError" not in result.stderr + + def test_error_names_the_offending_value(self, tmp_path: Path) -> None: + """Named in the one-line message, not merely in a traceback frame. + + Anchored to the start of a line: an uncaught raise renders the same + text under `ValueError: `, which *contains* `Error: ` as a substring, + so a plain `in` check would pass on the very traceback this guards + against. + """ + stderr = self._run(tmp_path, "1.0.0; id").stderr + + (error_line,) = [ + line for line in stderr.splitlines() if line.startswith("Error: ") + ] + assert "1.0.0; id" in error_line + + def test_nothing_is_installed_before_the_version_is_rejected( + self, tmp_path: Path + ) -> None: + """The banner prints, then it bails -- no download is announced.""" + result = self._run(tmp_path, "--help") + + assert result.returncode == 1 + assert "Downloading Claude Code CLI version" not in result.stdout diff --git a/tests/test_update_cli_version.py b/tests/test_update_cli_version.py new file mode 100644 index 000000000..a5d0c2641 --- /dev/null +++ b/tests/test_update_cli_version.py @@ -0,0 +1,335 @@ +"""Tests for scripts/update_cli_version.py version validation and file writing.""" + +import importlib.util +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).parent.parent +SCRIPT_PATH = REPO_ROOT / "scripts" / "update_cli_version.py" + +# scripts/ is not a package, so load update_cli_version.py by path +_spec = importlib.util.spec_from_file_location("update_cli_version", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +update_cli_version = importlib.util.module_from_spec(_spec) +sys.modules["update_cli_version"] = update_cli_version +_spec.loader.exec_module(update_cli_version) + +DEV_VERSION = "2.1.146-dev.20260519.t105443.shaece3dab" + +ORIGINAL = '"""Bundled Claude Code CLI version."""\n\n__cli_version__ = "2.1.195"\n' + + +@pytest.fixture +def version_file(tmp_path: Path) -> Path: + """A stand-in for src/claude_agent_sdk/_cli_version.py.""" + path = tmp_path / "_cli_version.py" + path.write_text(ORIGINAL) + return path + + +def import_version(path: Path) -> str: + """Import the written file as Python and return its __cli_version__. + + Fails on a file that is not valid Python, which is the whole point: the + value goes into a real source file that later gets imported. + """ + spec = importlib.util.spec_from_file_location(f"_written_{path.stem}", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + version: str = module.__cli_version__ + return version + + +class TestAcceptedVersions: + """Concrete versions round-trip through the file unchanged.""" + + @pytest.mark.parametrize( + "version", + [ + "1.2.3", + "2.1.195", + "2.1.201", + DEV_VERSION, + "1.2.3-beta.1", + "1.2.3-rc.1+build.4", + ], + ) + def test_round_trip(self, version_file: Path, version: str) -> None: + update_cli_version.update_cli_version(version, version_file) + assert import_version(version_file) == version + + @pytest.mark.parametrize( + ("argument", "written"), + [ + ("1.2.3\n", "1.2.3"), + ("1.2.3\r\n", "1.2.3"), + (" 2.1.201 ", "2.1.201"), + (f"{DEV_VERSION}\n", DEV_VERSION), + ], + ) + def test_surrounding_whitespace_is_stripped_before_writing( + self, version_file: Path, argument: str, written: str + ) -> None: + """The stripped value is written, so a trailing newline from a file read + never lands inside the string literal.""" + update_cli_version.update_cli_version(argument, version_file) + assert import_version(version_file) == written + + def test_docstring_and_shape_preserved(self, version_file: Path) -> None: + update_cli_version.update_cli_version(DEV_VERSION, version_file) + assert version_file.read_text() == ORIGINAL.replace("2.1.195", DEV_VERSION) + + def test_only_first_assignment_replaced(self, tmp_path: Path) -> None: + path = tmp_path / "_cli_version.py" + path.write_text('__cli_version__ = "1.0.0"\n_other = "1.0.0"\n') + update_cli_version.update_cli_version("2.0.0", path) + assert path.read_text() == '__cli_version__ = "2.0.0"\n_other = "1.0.0"\n' + + +class TestRejectedVersions: + """Invalid input raises before the file is touched.""" + + @pytest.mark.parametrize( + "version", + [ + # Quote breakout: closes the string literal in a real source file. + '1.0.0"', + '1.0.0" + __import__("os").system("id") + "', + '"', + # Backslashes: re.sub() replacement-escape processing, and invalid + # Python escapes in the emitted literal. + "1.0.0\\", + "\\", + "1.0.0\\1", + "1.0.0\\g<0>", + "\\g<0>", + "1.0.0\\n", + # Newlines *inside* the value: an extra source line in the + # generated module. (A merely trailing newline is stripped -- see + # TestAcceptedVersions.) + "1.0.0\nimport os", + "\n", + # Empty and flag-shaped. + "", + "-1.2.3", + "--help", + "-s", + # Leftovers from the shell-injection allowlist. + "1.0.0; id", + "$(id)", + "`id`", + "1.0.0 2.0.0", + " 1.0.0 2.0.0 ", + "../../etc/passwd", + ".1.2.3", + # Not versions at all: the old allowlist took these for concrete + # versions and pinned them into the file, deferring the failure to + # the wheel build. + "0", + "1.2", + "1.2.3.4", + "1.2.3+build.4", + "v2.1.207", + "next", + "beta", + # Case variants of the dist-tags: the installer's grammar is + # case-sensitive, so these are not tags it would resolve either. + "LATEST", + "Stable", + ], + ) + def test_rejected_and_file_untouched( + self, version_file: Path, version: str + ) -> None: + with pytest.raises(ValueError, match="Invalid CLI version"): + update_cli_version.update_cli_version(version, version_file) + assert version_file.read_text() == ORIGINAL + + @pytest.mark.parametrize("tag", ["latest", "stable"]) + def test_dist_tags_are_rejected(self, version_file: Path, tag: str) -> None: + """_cli_version.py must name one concrete build, never a moving tag. + + Five release runners each run build_wheel.py -> download_cli.py with + this value, so a dist-tag would let them resolve different CLI builds + into wheels published under one SDK version. "stable" is the dangerous + one: it passes the installer too, so before this guard it pinned a + moving tag and the build *succeeded*. + """ + with pytest.raises(ValueError, match="Invalid CLI version") as excinfo: + update_cli_version.update_cli_version(tag, version_file) + + assert "moving dist-tag" in str(excinfo.value) + assert "not a concrete version" in str(excinfo.value) + assert version_file.read_text() == ORIGINAL + + def test_leading_v_is_named_not_normalized(self, version_file: Path) -> None: + """It reached the file before, and failed at wheel-build time.""" + with pytest.raises(ValueError) as excinfo: + update_cli_version.update_cli_version("v2.1.207", version_file) + + assert "Did you mean '2.1.207'?" in str(excinfo.value) + assert version_file.read_text() == ORIGINAL + + def test_unsupported_dist_tag_is_named(self, version_file: Path) -> None: + with pytest.raises(ValueError, match="not a supported dist-tag"): + update_cli_version.update_cli_version("next", version_file) + assert version_file.read_text() == ORIGINAL + + def test_error_names_the_offending_value(self, version_file: Path) -> None: + with pytest.raises(ValueError) as excinfo: + update_cli_version.update_cli_version('1.0.0"; id', version_file) + assert '1.0.0"; id' in str(excinfo.value) + + def test_missing_assignment_leaves_file_untouched(self, tmp_path: Path) -> None: + path = tmp_path / "_cli_version.py" + path.write_text("# no assignment here\n") + with pytest.raises(ValueError, match="No __cli_version__ assignment"): + update_cli_version.update_cli_version("1.2.3", path) + assert path.read_text() == "# no assignment here\n" + + +class TestReplacementIsLiteral: + """The version reaches the file verbatim, with no escape interpretation. + + Validation already excludes every character these tests use, so they stub + it out to exercise the write path directly. Without that, a plain-string + re.sub() replacement -- which expands \\1 and \\g<0>, turns \\n into a + newline, and raises on a bare trailing backslash -- would look identical to + the callable replacement for all reachable input, and the guard would be + unfalsifiable. This is the test that fails if someone swaps the callable + back for an f-string. + """ + + @pytest.fixture + def no_validation(self, monkeypatch: pytest.MonkeyPatch) -> None: + def accept(version: str, *, source: str, allow_dist_tag: bool) -> str: + return version + + monkeypatch.setattr( + update_cli_version.version_validation, "validate_version", accept + ) + + @pytest.mark.usefixtures("no_validation") + @pytest.mark.parametrize( + "version", + [ + "2.0.0\\1", + "2.0.0\\g<0>", + "2.0.0\\", + "\\g<0>\\1", + '2.0.0"', + "2.0.0\\n", + "2.0.0\nid", + ], + ) + def test_written_file_imports_back_to_the_exact_string( + self, version_file: Path, version: str + ) -> None: + update_cli_version.update_cli_version(version, version_file) + assert import_version(version_file) == version + + @pytest.mark.usefixtures("no_validation") + def test_backreference_is_not_expanded_into_the_file( + self, version_file: Path + ) -> None: + """A string replacement would splice the matched assignment into itself.""" + update_cli_version.update_cli_version("2.0.0\\g<0>", version_file) + text = version_file.read_text() + assert text.count("__cli_version__") == 1 + + def test_replacement_is_a_callable( + self, monkeypatch: pytest.MonkeyPatch, version_file: Path + ) -> None: + """Guard the mechanism itself, not only its observable output.""" + spy = _SubnSpy(update_cli_version.ASSIGNMENT_PATTERN) + monkeypatch.setattr(update_cli_version, "ASSIGNMENT_PATTERN", spy) + + update_cli_version.update_cli_version("1.2.3", version_file) + + (repl,) = spy.replacements + assert callable(repl), ( + f"re.sub replacement must be a callable to avoid escape processing, " + f"got {repl!r}" + ) + + +class _SubnSpy: + """Records the replacement handed to Pattern.subn(), then delegates.""" + + def __init__(self, pattern: re.Pattern[str]) -> None: + self._pattern = pattern + self.replacements: list[object] = [] + + def subn(self, repl: object, string: str, count: int = 0) -> tuple[str, int]: + self.replacements.append(repl) + return self._pattern.subn(repl, string, count=count) # type: ignore[arg-type] + + +class TestCommandLine: + """The script still works when run directly, with scripts/ not a package. + + Every case runs in a throwaway cwd holding a copy of the real + src/claude_agent_sdk/_cli_version.py layout, so a regression in the guard + corrupts the fixture rather than the repository. + """ + + @pytest.fixture + def cwd(self, tmp_path: Path) -> Path: + target = tmp_path / "src" / "claude_agent_sdk" / "_cli_version.py" + target.parent.mkdir(parents=True) + target.write_text(ORIGINAL) + return tmp_path + + def _run(self, cwd: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(SCRIPT_PATH), *args], + cwd=cwd, + capture_output=True, + text=True, + ) + + def _target(self, cwd: Path) -> Path: + return cwd / "src" / "claude_agent_sdk" / "_cli_version.py" + + def test_invalid_version_exits_nonzero_without_writing(self, cwd: Path) -> None: + result = self._run(cwd, '9.9.9"; import os') + + assert result.returncode == 1 + assert "Invalid CLI version" in result.stderr + assert result.stdout == "" + assert self._target(cwd).read_text() == ORIGINAL + + @pytest.mark.parametrize("tag", ["latest", "stable"]) + def test_dist_tag_exits_nonzero_without_writing(self, cwd: Path, tag: str) -> None: + """`stable` used to pin cleanly and build cleanly -- against whatever + the tag happened to point at that day.""" + result = self._run(cwd, tag) + + assert result.returncode == 1 + assert "not a concrete version" in result.stderr + assert self._target(cwd).read_text() == ORIGINAL + + def test_leading_v_exits_nonzero_without_writing(self, cwd: Path) -> None: + result = self._run(cwd, "v2.1.207") + + assert result.returncode == 1 + assert "Did you mean '2.1.207'?" in result.stderr + assert self._target(cwd).read_text() == ORIGINAL + + def test_missing_argument_prints_usage_to_stderr(self, cwd: Path) -> None: + result = self._run(cwd) + assert result.returncode == 1 + assert "Usage:" in result.stderr + + def test_writes_when_run_directly(self, cwd: Path) -> None: + """Direct execution must resolve the shared validation module.""" + result = self._run(cwd, DEV_VERSION) + + assert result.returncode == 0, result.stderr + assert "ModuleNotFoundError" not in result.stderr + assert import_version(self._target(cwd)) == DEV_VERSION