Validate CLAUDE_CLI_VERSION and remove shell interpolation from the build scripts

.github/workflows/lint.yml

@@ -23,11 +23,11 @@
         python -m pip install --upgrade pip
         pip install -e ".[dev]"
 
     - name: Run ruff
       run: |
-        ruff check src/ tests/
-        ruff format --check src/ tests/
+        ruff check src/ tests/ scripts/
+        ruff format --check src/ tests/ scripts/
 
     - name: Run mypy
       run: |
-        mypy src/
+        mypy src/ scripts/

.github/workflows/test.yml

@@ -60,6 +60,10 @@ jobs:
 
       - name: Install Claude Code (Linux/macOS)
         if: runner.os == 'Linux' || runner.os == 'macOS'
+        # `shell: bash` runs with -eo pipefail. Without pipefail the step takes
+        # its status from `bash`, which exits 0 on an empty script, so a failed
+        # `curl` leaves the CLI uninstalled and the step green.
+        shell: bash
         run: |
           curl -fsSL https://claude.ai/install.sh | bash
           echo "$HOME/.local/bin" >> $GITHUB_PATH
@@ -149,6 +153,8 @@ jobs:
 
       - name: Install Claude Code (Linux)
         if: runner.os == 'Linux'
+        # `shell: bash` runs with -eo pipefail; see the note in the test job.
+        shell: bash
         run: |
           curl -fsSL https://claude.ai/install.sh | bash
           echo "$HOME/.local/bin" >> $GITHUB_PATH

pyproject.toml

@@ -101,6 +101,12 @@ strict_equality = true
 module = ["opentelemetry", "opentelemetry.*"]
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+# twine is not a declared dependency (not even in the [dev] extra); the
+# import in scripts/build_wheel.py is guarded by try/except ImportError.
+module = ["twine"]
+ignore_missing_imports = true
+
 [tool.ruff]
 target-version = "py310"
 line-length = 88

scripts/_cli_version_validation.py

@@ -0,0 +1,69 @@
+r"""Shared validation for Claude Code CLI version strings.
+
+Two scripts constrain the same value: update_cli_version.py writes it into
+src/claude_agent_sdk/_cli_version.py, and download_cli.py (reached from
+build_wheel.py) reads it back out and hands it to an installer. A second copy
+of the rule would let the writer emit a value the reader rejects, so the
+pattern and its validation helper live here once.
+
+A version must start with an alphanumeric character (so flag-shaped values
+like "--help" are rejected) and may then contain only characters that appear
+in real CLI versions, including dev builds such as
+"2.1.146-dev.20260519.t105443.shaece3dab".
+
+This allowlist is a security boundary, not just input hygiene:
+
+  * update_cli_version.update_cli_version() writes the version into a Python
+    string literal in a real source file, so it must never admit a double
+    quote, a backslash, or a newline.
+  * download_cli.download_cli() hands the version to an installer. Neither of
+    its paths interpolates it into a command string -- Unix passes it as its
+    own argv element, Windows passes it in the environment -- so for that
+    caller the allowlist is defense in depth rather than the only barrier.
+
+Widening it requires re-reading tests/test_download_cli.py::TestGetCliVersion
+and tests/test_update_cli_version.py.
+
+Deliberately unanchored, and matched with fullmatch() rather than match():
+with "^...$" a swap to match() would silently accept a trailing newline
+("1.0.0\n"); unanchored, the same swap accepts obvious prefixes like
+"1.0.0; id" and fails immediately in tests.
+"""
+
+import re
+
+VERSION_PATTERN = re.compile(r"[0-9A-Za-z][0-9A-Za-z.+-]*")
+
+
+def validate_version(version: str, *, source: str, allow_latest: bool) -> str:
+    """Return ``version`` unchanged if it is a usable CLI version.
+
+    Args:
+        version: The candidate version string.
+        source: Name of where the value came from, used in the error message
+            (e.g. "CLAUDE_CLI_VERSION").
+        allow_latest: Whether the "latest" sentinel is acceptable. It is for a
+            download, which resolves it at install time; it is not for a value
+            pinned into _cli_version.py, which must name one concrete build.
+
+    Raises:
+        ValueError: If ``version`` is neither an allowed "latest" nor a
+            fullmatch of VERSION_PATTERN.
+    """
+    # "latest" fullmatches VERSION_PATTERN, so it has to be ruled out by name
+    # rather than by the pattern when it is not allowed.
+    if version == "latest":
+        if allow_latest:
+            return version
+        raise ValueError(
+            f"Invalid {source}: 'latest' is not a concrete version. "
+            f"Expected a version matching {VERSION_PATTERN.pattern}"
+        )
+
+    if not VERSION_PATTERN.fullmatch(version):
+        expected = "'latest' or a version" if allow_latest else "a concrete version"
+        raise ValueError(
+            f"Invalid {source}: {version!r}. "
+            f"Expected {expected} matching {VERSION_PATTERN.pattern}"
+        )
+    return version

scripts/check_pypi_quota.py

@@ -15,12 +15,13 @@
 import sys
 import urllib.request
 from pathlib import Path
+from typing import Any, cast
 
 PYPI_PROJECT_LIMIT_BYTES = 50 * 1024**3  # 50 GiB (increased from PyPI default 10 GiB)
 PYPI_FILE_LIMIT_BYTES = 100 * 1024**2  # 100 MiB
 
 
-def fetch_project_files(package: str) -> list[dict]:
+def fetch_project_files(package: str) -> list[dict[str, Any]]:
     req = urllib.request.Request(
         f"https://pypi.org/simple/{package}/",
         headers={
@@ -30,15 +31,16 @@ def fetch_project_files(package: str) -> list[dict]:
     )
     with urllib.request.urlopen(req, timeout=30) as resp:
         data = json.load(resp)
-    return data.get("files", [])
+    return cast("list[dict[str, Any]]", data.get("files", []))
 
 
 def human(n: int) -> str:
+    size = float(n)
     for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
-        if abs(n) < 1024 or unit == "TiB":
-            return f"{n:.2f} {unit}"
-        n /= 1024
-    return f"{n:.2f} TiB"
+        if abs(size) < 1024 or unit == "TiB":
+            return f"{size:.2f} {unit}"
+        size /= 1024
+    return f"{size:.2f} TiB"
 
 
 def main() -> int:

scripts/download_cli.py

@@ -11,13 +11,43 @@
 import shutil
 import subprocess
 import sys
+import tempfile
 import time
+from collections.abc import Callable
 from pathlib import Path
 
+# scripts/ is not a package. Running this file directly -- as build_wheel.py
+# does, via `python scripts/download_cli.py` -- already puts scripts/ on
+# sys.path, but loading it by path (importlib.spec_from_file_location, as the
+# tests do) does not. Add it either way so the shared module resolves.
+_SCRIPTS_DIR = str(Path(__file__).parent)
+if _SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, _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 out of the environment under this
+# name instead of having it spliced into the PowerShell command text.
+INSTALL_VERSION_ENV_VAR = "CLAUDE_CLI_INSTALL_VERSION"
+
+# How many times retry_install() runs an attempt before giving up.
+MAX_INSTALL_ATTEMPTS = 3
+
 
 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.
+
+    Raises:
+        ValueError: If CLAUDE_CLI_VERSION is set to something other than
+            "latest" or a value matching VERSION_PATTERN.
+    """
+    version = os.environ.get("CLAUDE_CLI_VERSION", "latest")
+    return version_validation.validate_version(
+        version, source="CLAUDE_CLI_VERSION", allow_latest=True
+    )
 
 
 def find_installed_cli() -> Path | None:
@@ -50,6 +80,75 @@
     return None
 
 
+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 check_install_script(script_path: str) -> None:
+    """Reject a downloaded install script 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"#!":
+        print(
+            f"Error: downloaded install script does not start with a shebang "
+            f"(first bytes: {magic!r}). Refusing to execute it.",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+
+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))
+
+    # The failure is reported from inside the handler for the last attempt, so
+    # the error is in scope and non-None by construction. Stashing it in a
+    # `CalledProcessError | None` and reading it after the loop instead would
+    # rely on the reader -- and the type checker -- knowing that range(1, 4)
+    # cannot be empty.
+    for attempt_num in range(1, MAX_INSTALL_ATTEMPTS + 1):
+        try:
+            attempt()
+            return
+        except subprocess.CalledProcessError as e:
+            if attempt_num == MAX_INSTALL_ATTEMPTS:
+                print(
+                    f"Error downloading CLI after {MAX_INSTALL_ATTEMPTS} attempts: {e}",
+                    file=sys.stderr,
+                )
+                print(f"stdout: {e.stdout.decode()}", file=sys.stderr)
+                print(f"stderr: {e.stderr.decode()}", 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)
+
+
 def download_cli() -> None:
     """Download Claude Code CLI using the official install script."""
     version = get_cli_version()
@@ -59,7 +158,13 @@
 
     # Build install command based on platform
     if system == "Windows":
-        # Use PowerShell installer on Windows
+        # Use PowerShell installer on Windows. The version is handed to
+        # PowerShell in the environment and referenced by name, so it is never
+        # part of the command text that PowerShell parses. `$env:NAME` in
+        # argument position expands to exactly one argument -- PowerShell does
+        # not re-split or re-parse it -- which is the argv separation the Unix
+        # path gets from `["bash", script, version]`.
+        install_env: dict[str, str] | None = None
         if version == "latest":
             install_cmd = [
                 "powershell",
@@ -74,39 +179,45 @@
                 "-ExecutionPolicy",
                 "Bypass",
                 "-Command",
-                f"& ([scriptblock]::Create((irm https://claude.ai/install.ps1))) {version}",
+                "& ([scriptblock]::Create((irm https://claude.ai/install.ps1))) "
+                f"$env:{INSTALL_VERSION_ENV_VAR}",
             ]
-    else:
+            install_env = {**os.environ, INSTALL_VERSION_ENV_VAR: version}
+        retry_install(lambda: run_command(install_cmd, env=install_env))
+        return
+
+    # Download install.sh to a file and run it directly rather than piping
+    # curl into bash through a shell string: nothing is interpolated into a
+    # command line, and check=True sees curl's and bash's exit codes
+    # separately instead of only the last status of a pipeline.
+    with tempfile.TemporaryDirectory() as tmpdir:
+        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. 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}"]
-
-    # Small jitter to stagger parallel matrix builds hitting the same endpoint
-    time.sleep(random.uniform(0, 5))
+        # 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 != "latest":
+            bash_cmd.append(version)
 
-    last_err: subprocess.CalledProcessError | None = None
-    for attempt in range(1, 4):
-        try:
-            subprocess.run(install_cmd, check=True, capture_output=True)
-            return
-        except subprocess.CalledProcessError as e:
-            last_err = e
-            if attempt < 3:
-                delay = 2**attempt
-                print(
-                    f"Install attempt {attempt} failed (exit {e.returncode}), "
-                    f"retrying in {delay}s...",
-                    file=sys.stderr,
-                )
-                time.sleep(delay)
+        def attempt() -> None:
+            run_command(curl_cmd)
+            check_install_script(script_path)
+            run_command(bash_cmd)
 
-    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)
+        retry_install(attempt)
 
 
 def copy_cli_to_bundle() -> None:
@@ -165,4 +276,14 @@
 
 
 if __name__ == "__main__":
-    main()
+    # get_cli_version() raises on a bad CLAUDE_CLI_VERSION, and this script
+    # runs as a build step. Report that 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, because update_cli_version.py
+    # and the tests read the exception; only the entry point turns it into an
+    # exit status. Matches update_cli_version.py's __main__.
+    try:
+        main()
+    except ValueError as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        sys.exit(1)