Single-quote POSIX shell tokens, share the tool probe, harden registry checks

shell_token now picks the quoting style per platform: single quotes on
POSIX (sh expands nothing inside them, so backslash runs, $VAR, and
backticks reach the compiler exactly as lexed, matching SCons's
no-shell spawn) and the CreateProcess argv rule on Windows. A test
round-trips every case through a real /bin/sh.

The ninja and ccache runnability probes collapse into one
tool_version_runs helper in framework_helpers. The registry names a
non-dict top-level payload like the inner guards, and the expect layout
check also runs on marker hits so a marked install that later lost
files fails by name. The ESP-IDF ccache gate defers to
resolve_ccache_path so ESPHOME_CCACHE_ENABLE=0 disables ccache there
too, with IDF_CCACHE_ENABLE still taking precedence.

esphome/build_helpers/ccache.py

@@ -13,40 +13,18 @@ from __future__ import annotations
 import logging
 import os
 from pathlib import Path
-import subprocess
 
-from esphome.framework_helpers import strip_win_long_path_prefix
+from esphome.framework_helpers import strip_win_long_path_prefix, tool_version_runs
 
 _LOGGER = logging.getLogger(__name__)
 
 
 def _ccache_runs(ccache: str) -> bool:
-    """Return True when the ``ccache`` found on PATH actually runs.
-
-    ``shutil.which`` proves existence, not runnability: on Windows it also
-    matches ``.bat``/``.cmd`` wrappers and stale package-manager shims whose
-    target is gone. Wrapping compiles around such a find fails every compile
-    step with an opaque OS error, so probe once and fall back to compiling
-    without ccache when the probe fails.
-    """
-    try:
-        subprocess.run(
-            [ccache, "--version"],
-            check=True,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            timeout=15,
-            # Repo-wide convention (posix_spawn fast path); see the
-            # close_fds=False call sites across esphome/ and script/helpers.py
-            close_fds=False,
-        )
-    except (OSError, subprocess.SubprocessError):
-        _LOGGER.warning(
-            "Ignoring ccache at %s because it failed to run; compiling without ccache",
-            ccache,
-        )
-        return False
-    return True
+    """Return True when the ``ccache`` found on PATH actually runs."""
+    return tool_version_runs(
+        ccache,
+        "Ignoring ccache at %s because it failed to run; compiling without ccache",
+    )
 
 
 def resolve_ccache_path() -> str | None:

esphome/build_helpers/ninja.py

@@ -7,38 +7,20 @@ import os
 from pathlib import Path
 import re
 import shutil
-import subprocess
 
 from esphome.core import EsphomeError
-from esphome.framework_helpers import strip_win_long_path_prefix
+from esphome.framework_helpers import strip_win_long_path_prefix, tool_version_runs
 
 _LOGGER = logging.getLogger(__name__)
 
 
 def _ninja_runs(binary: str) -> bool:
-    """Whether the ninja found on PATH actually runs.
-
-    Same rationale as the ccache probe: ``shutil.which`` proves existence,
-    not runnability (stale shims, broken wrappers).
-    """
-    try:
-        subprocess.run(
-            [binary, "--version"],
-            check=True,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            timeout=15,
-            # Repo-wide convention (posix_spawn fast path)
-            close_fds=False,
-        )
-    except (OSError, subprocess.SubprocessError):
-        _LOGGER.warning(
-            "Ignoring ninja at %s because it failed to run; "
-            "falling back to the bundled wheel",
-            binary,
-        )
-        return False
-    return True
+    """Whether the ninja found on PATH actually runs (see tool_version_runs)."""
+    return tool_version_runs(
+        binary,
+        "Ignoring ninja at %s because it failed to run; "
+        "falling back to the bundled wheel",
+    )
 
 
 def find_ninja() -> Path:
@@ -80,8 +62,9 @@ def quote_arg(tok: str) -> str:
 
     Same escaping rule as ``subprocess.list2cmdline``: a backslash run
     doubles only immediately before a quote (or the closing quote), and the
-    quote itself is escaped. POSIX sh parses the result identically for
-    backslashes and quotes. ``$`` must already be doubled for ninja.
+    quote itself is escaped. CreateProcess-only; POSIX sh collapses
+    backslash runs inside double quotes, so shell_token single-quotes
+    there instead. ``$`` must already be doubled for ninja.
     """
     quoted = re.sub(r'(\\*)"', lambda m: m.group(1) * 2 + '\\"', tok)
     quoted = re.sub(r"(\\+)\Z", lambda m: m.group(1) * 2, quoted)
@@ -99,17 +82,22 @@ def shell_token(tok: str, force: bool = False) -> str:
 
     Lexing strips the quoting a user wrote (``-DX="a b"`` becomes the single
     token ``-DX=a b``); re-quote on the way out so the compiler receives the
-    same argv element SCons would pass under PlatformIO. After ninja
-    un-doubles ``$$``, sh still applies every expansion double quotes allow
-    (``$VAR``, ``$(...)``, backticks) while CreateProcess passes them
-    literally; SCons on POSIX spawns without a shell, so this is a known,
-    deliberate divergence for tokens carrying those characters.
+    same argv element SCons would pass under PlatformIO. Ninja hands POSIX
+    commands to ``/bin/sh -c`` and Windows commands to CreateProcess, so the
+    quoting style is chosen per platform: single quotes on POSIX (sh expands
+    nothing inside them, matching SCons's no-shell spawn) and the argv rule
+    on Windows. ``$`` is doubled first in either case because ninja expands
+    ``$`` before the command reaches the shell.
     """
     tok = tok.replace("$", "$$")  # ninja would expand a bare $ to nothing
-    if force or not tok or _NEEDS_QUOTE.search(tok):
-        # An empty token must become "" or it vanishes from the argv
+    if not (force or not tok or _NEEDS_QUOTE.search(tok)):
+        return tok
+    # An empty token must become '' / "" or it vanishes from the argv
+    if os.name == "nt":
         return quote_arg(tok)
-    return tok
+    # shlex.quote's rule; inlined because the $-doubled token must not be
+    # re-examined for safe characters
+    return "'" + tok.replace("'", "'\"'\"'") + "'"
 
 
 def quote_path(value: Path | str) -> str:

esphome/espidf/framework.py

@@ -11,7 +11,7 @@ import re
 import shutil
 from typing import Any, NoReturn
 
-from esphome.build_helpers.ccache import ccache_defaults_env
+from esphome.build_helpers.ccache import ccache_defaults_env, resolve_ccache_path
 from esphome.build_helpers.tools_cache import tools_cache_path
 from esphome.core import Version
 from esphome.framework_helpers import (
@@ -1150,10 +1150,14 @@ def _ccache_env() -> dict[str, str]:
     a custom ``CCACHE_DIR`` / ``CCACHE_MAXSIZE`` / etc. is respected.
     """
     # Honor an explicit choice already in the environment (opt-out or opt-in).
+    # IDF_CCACHE_ENABLE (this backend's native knob) wins over the shared
+    # ESPHOME_CCACHE_ENABLE, which resolve_ccache_path parses; without it a
+    # user disabling ccache to debug a miscompile would silently keep it
+    # enabled here.
     if "IDF_CCACHE_ENABLE" in os.environ:
         if not get_bool_env("IDF_CCACHE_ENABLE"):
             return {}
-    elif shutil.which("ccache") is None:
+    elif resolve_ccache_path() is None:
         # ESP-IDF silently skips ccache without the binary; don't enable it.
         return {}
 

esphome/framework_helpers.py

@@ -196,6 +196,30 @@ def run_command(
         return False, None, None
 
 
+def tool_version_runs(binary: str, warning: str) -> bool:
+    """Probe ``binary --version``; on failure warn with ``warning`` % binary.
+
+    ``shutil.which`` proves existence, not runnability: on Windows it also
+    matches ``.bat``/``.cmd`` wrappers and stale package-manager shims whose
+    target is gone. Callers probe once and fall back instead of failing
+    every build step with an opaque OS error.
+    """
+    try:
+        subprocess.run(
+            [binary, "--version"],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=15,
+            # Repo-wide convention (posix_spawn fast path)
+            close_fds=False,
+        )
+    except (OSError, subprocess.SubprocessError):
+        _LOGGER.warning(warning, binary)
+        return False
+    return True
+
+
 def run_command_ok(*args, **kwargs) -> bool:
     """
     Execute a command and return only the success status.

esphome/platformio/registry.py

@@ -70,6 +70,10 @@ def registry_download(package: str, version: str) -> tuple[str, str, int | None]
         raise EsphomeError(
             f"The package registry returned invalid JSON for {package}: {err}"
         ) from err
+    if not isinstance(data, dict):
+        raise EsphomeError(
+            f"Unexpected package registry response for {package}: {str(data)[:200]}"
+        )
     systype = get_systype()
     versions = data.get("versions")
     if not isinstance(versions, list):
@@ -127,6 +131,21 @@ def registry_download(package: str, version: str) -> tuple[str, str, int | None]
     raise EsphomeError(f"{package} {version} not found in the package registry")
 
 
+def _check_layout(name: str, dest: Path, expect: Collection[str]) -> None:
+    """Raise when an install tree is missing an expected directory.
+
+    Runs on fresh extracts and on marker hits: a marked tree that later
+    lost files (manual deletion, antivirus quarantine) must fail by name
+    instead of surfacing as an opaque toolchain error.
+    """
+    for rel in expect:
+        if not (dest / rel).is_dir():
+            raise EsphomeError(
+                f"{name} at {dest} is missing the expected {rel} "
+                "directory; run 'esphome clean-all' and retry"
+            )
+
+
 def install_package(
     name: str,
     version: str,
@@ -148,6 +167,7 @@ def install_package(
         raise ValueError("install_package requires a non-empty expect")
     marker = dest / ".esphome_extracted"
     if marker.is_file():
+        _check_layout(name, dest, expect)
         return
     from filelock import FileLock
 
@@ -185,11 +205,6 @@ def install_package(
         archive_extract_all(archive, dest, progress_header="Extracting")
         # Validate the layout before recording success, so an unexpected
         # package is never cached as a working install.
-        for rel in expect:
-            if not (dest / rel).is_dir():
-                raise EsphomeError(
-                    f"{name} {version} extracted without the expected {rel} "
-                    "directory; run 'esphome clean-all' and retry"
-                )
+        _check_layout(name, dest, expect)
         marker.touch()
         archive.unlink(missing_ok=True)

tests/unit_tests/build_helpers/test_ccache.py

@@ -30,9 +30,7 @@ def test_resolve_probe_failure() -> None:
     with (
         patch.dict(os.environ, {}, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch(
-            "esphome.build_helpers.ccache.subprocess.run", side_effect=OSError("boom")
-        ),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")),
     ):
         assert ccache.resolve_ccache_path() is None
 
@@ -55,7 +53,7 @@ def test_resolve_explicit_skips_probe_and_warns_missing(
 
 
 def test_probe_spawns_with_close_fds_false() -> None:
-    with patch("esphome.build_helpers.ccache.subprocess.run") as mock_run:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
         assert ccache._ccache_runs("/usr/bin/ccache") is True
     assert mock_run.call_args.kwargs["close_fds"] is False
 

tests/unit_tests/build_helpers/test_ninja.py

@@ -57,6 +57,11 @@ def test_escape_ninja_specials() -> None:
     assert ninja_helper.escape("a b:c$d") == "a$ b$:c$$d"
 
 
+def _q(tok: str) -> str:
+    """The platform's shell_token quote wrapper (argv rule on Windows)."""
+    return f'"{tok}"' if os.name == "nt" else f"'{tok}'"
+
+
 def test_quote_arg_windows_argv_rule() -> None:
     # Backslash runs double only before a quote (subprocess.list2cmdline rule)
     assert ninja_helper.quote_arg('-DX=a\\"b c') == '"-DX=a\\\\\\"b c"'
@@ -65,25 +70,43 @@ def test_quote_arg_windows_argv_rule() -> None:
 
 def test_shell_token_quotes_only_when_needed() -> None:
     assert ninja_helper.shell_token("-Os") == "-Os"
-    assert ninja_helper.shell_token("-DP=C:\\x y") == '"-DP=C:\\x y"'
-    assert ninja_helper.shell_token("plain", force=True) == '"plain"'
+    assert ninja_helper.shell_token("-DP=C:\\x y") == _q("-DP=C:\\x y")
+    assert ninja_helper.shell_token("plain", force=True) == _q("plain")
 
 
 def test_shell_token_quotes_shell_metacharacters() -> None:
     """Tokens like -DMASK=(1<<3) must not reach /bin/sh -c bare."""
-    assert ninja_helper.shell_token("-DMASK=(1<<3)") == '"-DMASK=(1<<3)"'
-    assert ninja_helper.shell_token("-DX=a;b") == '"-DX=a;b"'
-    assert ninja_helper.shell_token("-DX=$HOME") == '"-DX=$$HOME"'
+    assert ninja_helper.shell_token("-DMASK=(1<<3)") == _q("-DMASK=(1<<3)")
+    assert ninja_helper.shell_token("-DX=a;b") == _q("-DX=a;b")
+    assert ninja_helper.shell_token("-DX=$HOME") == _q("-DX=$$HOME")
+
+
+def test_shell_token_posix_roundtrips_through_sh() -> None:
+    """Backslash runs, $, backticks, and quotes must reach the compiler
+    exactly as lexed once ninja un-doubles $$ and /bin/sh strips quotes."""
+    import subprocess
+
+    if sys.platform == "win32":
+        pytest.skip("POSIX sh quoting")
+    for tok in ("-DP=a\\\\b", "-DX=$VAR", "-DY=`date`", "-DZ=it's", '-DC="q"'):
+        quoted = ninja_helper.shell_token(tok).replace("$$", "$")
+        out = subprocess.run(
+            ["/bin/sh", "-c", f'printf "%s" {quoted}'],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        assert out.stdout == tok
 
 
 def test_quote_path_force_quotes() -> None:
-    assert ninja_helper.quote_path(Path("a b")) == '"a b"'
-    assert ninja_helper.quote_path("simple") == '"simple"'
+    assert ninja_helper.quote_path(Path("a b")) == _q("a b")
+    assert ninja_helper.quote_path("simple") == _q("simple")
 
 
 def test_shell_token_empty_token_is_quoted() -> None:
     """An empty argv element must survive as an explicit pair of quotes."""
-    assert ninja_helper.shell_token("") == '""'
+    assert ninja_helper.shell_token("") == _q("")
 
 
 def test_find_ninja_probes_path_hit(tmp_path: Path) -> None:
@@ -101,14 +124,12 @@ def test_find_ninja_probes_path_hit(tmp_path: Path) -> None:
 
 
 def test_ninja_probe_failure_warns(caplog: pytest.LogCaptureFixture) -> None:
-    with patch(
-        "esphome.build_helpers.ninja.subprocess.run", side_effect=OSError("boom")
-    ):
+    with patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")):
         assert ninja_helper._ninja_runs("/broken/ninja") is False
     assert "failed to run" in caplog.text
 
 
 def test_ninja_probe_success() -> None:
-    with patch("esphome.build_helpers.ninja.subprocess.run") as mock_run:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
         assert ninja_helper._ninja_runs("/usr/bin/ninja") is True
     assert mock_run.call_args.kwargs["close_fds"] is False

tests/unit_tests/test_espidf_framework.py

@@ -1395,7 +1395,9 @@ def test_get_framework_env_without_python_env_uses_os_path(tmp_path: Path) -> No
 
 def _ccache_patches(tmp_path: Path, which: str | None, build_path: Path | None):
     return (
-        patch("esphome.espidf.framework.shutil.which", return_value=which),
+        # The gate defers to the shared resolver (which carries the PATH
+        # lookup, ESPHOME_CCACHE_ENABLE parse, and runnability probe)
+        patch("esphome.espidf.framework.resolve_ccache_path", return_value=which),
         patch(
             "esphome.espidf.framework.get_idf_tools_path",
             return_value=tmp_path / "tools",
@@ -1445,6 +1447,26 @@ def test_ccache_env_opt_in_without_binary(tmp_path: Path) -> None:
     assert env["CCACHE_DEPEND"] == "1"
 
 
+def test_ccache_env_honors_shared_esphome_opt_out(tmp_path: Path) -> None:
+    """ESPHOME_CCACHE_ENABLE=0 disables ccache here too; the shared policy
+    must not apply to every backend except this one."""
+    _p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    env_vars = {"ESPHOME_CCACHE_ENABLE": "0", "PATH": "/usr/bin"}
+    with patch.dict("os.environ", env_vars, clear=True), p2, p3:
+        # The real resolver runs so the opt-out parse is exercised
+        assert _ccache_env() == {}
+
+
+def test_ccache_env_idf_knob_wins_over_shared_opt_out(tmp_path: Path) -> None:
+    """IDF_CCACHE_ENABLE=1 takes precedence over ESPHOME_CCACHE_ENABLE=0."""
+    p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
+    env_vars = {"IDF_CCACHE_ENABLE": "1", "ESPHOME_CCACHE_ENABLE": "0"}
+    with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
+        env = _ccache_env()
+    assert env["CCACHE_DIR"] == str(tmp_path / "tools" / "ccache")
+    assert "IDF_CCACHE_ENABLE" not in env
+
+
 def test_ccache_env_preserves_user_overrides(tmp_path: Path) -> None:
     # User-set CCACHE_* values must not be clobbered; unset ones still default.
     p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")

tests/unit_tests/test_platformio_registry.py

@@ -201,7 +201,7 @@ def test_registry_download_version_not_found() -> None:
 
 def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
     dest = tmp_path / "pkg"
-    dest.mkdir()
+    (dest / "payload").mkdir(parents=True)
     (dest / ".esphome_extracted").touch()
     with patch.object(registry, "download_from_mirrors") as mock_download:
         registry.install_package(
@@ -210,6 +210,18 @@ def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
     mock_download.assert_not_called()
 
 
+def test_install_package_marker_hit_rechecks_layout(tmp_path: Path) -> None:
+    """A marked install that later lost files fails by name instead of
+    surfacing as an opaque toolchain error."""
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    (dest / ".esphome_extracted").touch()
+    with pytest.raises(EsphomeError, match="missing the expected payload"):
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+
+
 def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
     dest = tmp_path / "pkg"
     mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
@@ -276,7 +288,7 @@ def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
         patch.object(registry, "download_from_mirrors"),
         patch.object(registry, "archive_extract_all") as mock_extract,
         patch.object(registry, "get_systype", return_value="linux_x86_64"),
-        pytest.raises(EsphomeError, match="without the expected bin"),
+        pytest.raises(EsphomeError, match="missing the expected bin"),
     ):
         mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
             parents=True
@@ -416,3 +428,17 @@ def test_registry_download_non_dict_file_entry_is_named() -> None:
         pytest.raises(EsphomeError, match="Unexpected package registry response"),
     ):
         registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_dict_payload_is_named() -> None:
+    """A JSON array answer is an unexpected payload at the outermost level."""
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps(["1.0.0"]).encode())
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")

tests/unit_tests/test_platformio_toolchain.py

@@ -432,7 +432,7 @@ def test_ccache_env_enabled_by_default(setup_core: Path) -> None:
     with (
         patch.dict(os.environ, {}, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -495,7 +495,7 @@ def test_ccache_env_disabled_when_probe_fails(
     with (
         patch.dict(os.environ, {}, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run", side_effect=probe_error),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=probe_error),
     ):
         env = toolchain._ccache_env()
 
@@ -509,7 +509,7 @@ def test_ccache_env_forced_on_skips_probe(setup_core: Path) -> None:
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run") as mock_probe,
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -539,7 +539,7 @@ def test_ccache_env_strips_win_long_path_prefix(setup_core: Path) -> None:
         # implementation (which crashes on a POSIX host) is never reached.
         patch("esphome.framework_helpers.sys.platform", "win32"),
         patch("shutil.which", return_value=prefixed),
-        patch("esphome.build_helpers.ccache.subprocess.run") as mock_probe,
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -588,7 +588,7 @@ def test_ccache_env_respects_user_values_and_refreshes_basedir(
     with (
         patch.dict(os.environ, user_env, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -607,7 +607,7 @@ def test_run_platformio_cli_passes_ccache_env_to_subprocess_only(
     with (
         patch.dict(os.environ, {}, clear=False),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         mock_run_external_process.return_value = 0
@@ -629,7 +629,7 @@ def test_ccache_env_requires_build_path(setup_core: Path) -> None:
     with (
         patch.dict(os.environ, {}, clear=True),
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run"),
+        patch("esphome.framework_helpers.subprocess.run"),
         pytest.raises(ValueError, match="CORE.build_path must be set"),
     ):
         toolchain._ccache_env()
@@ -643,7 +643,7 @@ def test_run_platformio_cli_merges_caller_env(
 
     with (
         patch("shutil.which", return_value="/usr/bin/ccache"),
-        patch("esphome.build_helpers.ccache.subprocess.run"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         mock_run_external_process.return_value = 0
         toolchain.run_platformio_cli(