From d614e9175365606bd6b7df1a0b72049ec4802316 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 20 Aug 2026 09:38:44 -0500 Subject: [PATCH] Cross-platform token quoting, ninja-safe dollar escaping, and a tighter install path --- esphome/arduino8266/framework.py | 77 ++++++++----------- esphome/build_gen/arduino8266.py | 53 ++++++++----- .../unit_tests/build_gen/test_arduino8266.py | 33 ++++++-- .../unit_tests/test_arduino8266_framework.py | 24 +++--- 4 files changed, 106 insertions(+), 81 deletions(-) diff --git a/esphome/arduino8266/framework.py b/esphome/arduino8266/framework.py index d008066729..066667806c 100644 --- a/esphome/arduino8266/framework.py +++ b/esphome/arduino8266/framework.py @@ -196,50 +196,39 @@ def _install_package( # filelock pattern as platformio/toolchain.py and git.py). dest.parent.mkdir(parents=True, exist_ok=True) with FileLock(f"{dest}.lock"): - _install_package_locked(name, version, dest, mirrors, expect, marker) - - -def _install_package_locked( - name: str, - version: str, - dest: Path, - mirrors: list[str], - expect: Collection[str], - marker: Path, -) -> None: - if marker.is_file(): - # Another process finished the install while we waited for the lock - return - rmdir(dest, msg=f"Clean up incomplete {name} install") - # A persistent download location (not a temp dir) so an interrupted - # download resumes across esphome runs via download_with_resume's .part - # file, mirroring the espidf dist/ convention. - archive = _downloads_path() / f"{name}-{version}" - _LOGGER.info("Downloading %s %s ...", name, version) - if mirrors: - _LOGGER.warning( - "Downloading %s from a mirror override; checksum verification " - "is skipped for mirrors", - name, - ) - download_from_mirrors( - mirrors, {"VERSION": version, "SYSTEM": _pio_system()}, archive - ) - else: - url, sha256, size = _registry_download(name, version) - download_with_resume(url, archive, sha256=sha256, size=size) - _LOGGER.info("Extracting %s ...", name) - 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" + if marker.is_file(): + # Another process finished the install while we waited + return + rmdir(dest, msg=f"Clean up incomplete {name} install") + # A persistent download location (not a temp dir) so an interrupted + # download resumes across esphome runs via download_with_resume's + # .part file, mirroring the espidf dist/ convention. + archive = _downloads_path() / f"{name}-{version}" + _LOGGER.info("Downloading %s %s ...", name, version) + if mirrors: + _LOGGER.warning( + "Downloading %s from a mirror override; checksum verification " + "is skipped for mirrors", + name, ) - marker.touch() - archive.unlink(missing_ok=True) + download_from_mirrors( + mirrors, {"VERSION": version, "SYSTEM": _pio_system()}, archive + ) + else: + url, sha256, size = _registry_download(name, version) + download_with_resume(url, archive, sha256=sha256, size=size) + _LOGGER.info("Extracting %s ...", name) + 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" + ) + marker.touch() + archive.unlink(missing_ok=True) def _find_ninja() -> Path: @@ -276,7 +265,7 @@ def check_and_install(framework_version: cv.Version) -> dict[str, Path]: package_version, framework_path, ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS, - expect=("cores/esp8266", "tools/sdk"), + expect=("cores/esp8266", "tools/sdk", "libraries"), ) toolchain_path = get_toolchain_path() _install_package( diff --git a/esphome/build_gen/arduino8266.py b/esphome/build_gen/arduino8266.py index 9e45598063..31371c0f18 100644 --- a/esphome/build_gen/arduino8266.py +++ b/esphome/build_gen/arduino8266.py @@ -270,8 +270,32 @@ def _e(value) -> str: def _q(value) -> str: - """Quote a path for use inside a ninja command line (shell/CreateProcess).""" - return f'"{value}"' + """Quote a path for use inside a ninja command line (shell/CreateProcess). + + ``$`` doubles so ninja passes it through literally instead of expanding + an (empty) ninja variable. + """ + return '"' + str(value).replace("$", "$$") + '"' + + +_NEEDS_QUOTE = re.compile(r'[\s"\']') + + +def _shell_token(tok: str) -> str: + """Quote a lexed token for the ninja command line; ``_q`` is for paths. + + 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. Uses the Windows + argv quoting rule, which POSIX sh parses identically inside double + quotes: a backslash run doubles only immediately before a quote. + """ + tok = tok.replace("$", "$$") # ninja would expand a bare $ to nothing + if not _NEEDS_QUOTE.search(tok): + return tok + quoted = re.sub(r'(\\*)"', lambda m: m.group(1) * 2 + '\\"', tok) + quoted = re.sub(r"(\\+)\Z", lambda m: m.group(1) * 2, quoted) + return f'"{quoted}"' def _defines_flags( @@ -308,19 +332,9 @@ def _unflag_tokens() -> set[str]: } -def _shell_token(tok: str) -> str: - """Quote a lexed token for the shell-expanded ninja command line. - - 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. - """ - if not re.search(r'[\s"\']', tok): - return tok - return '"' + tok.replace("\\", "\\\\").replace('"', '\\"') + '"' - - -def _project_flags() -> tuple[list[str], list[str], list[Path], list[str]]: +def _project_flags( + unflags: set[str], +) -> tuple[list[str], list[str], list[Path], list[str]]: """Split the ESPHome build flags into compile, linker, -L, and -l lists. Every entry is shell-lexed the way PlatformIO's ``ParseFlags`` does, so a @@ -328,7 +342,6 @@ def _project_flags() -> tuple[list[str], list[str], list[Path], list[str]]: ``build_unflags`` matches individual tokens (``-Os`` inside ``-Os -g3``). Lexed tokens are re-quoted at emission via ``_shell_token``. """ - unflags = _unflag_tokens() compile_flags: list[str] = [] link_flags: list[str] = [] lib_dirs: list[Path] = [] @@ -338,14 +351,14 @@ def _project_flags() -> tuple[list[str], list[str], list[Path], list[str]]: if tok in unflags: continue if tok.startswith("-Wl,"): - link_flags.append(tok) + link_flags.append(_shell_token(tok)) elif tok.startswith("-L"): lib_dirs.append(Path(tok[2:])) elif tok.startswith("-l"): libs.append(tok[2:]) else: compile_flags.append(_shell_token(tok)) - return compile_flags, [_shell_token(t) for t in link_flags], lib_dirs, libs + return compile_flags, link_flags, lib_dirs, libs def _collect_sources(root: Path, exclude: set[str] = frozenset()) -> list[Path]: @@ -501,12 +514,13 @@ def write_project(paths: dict[str, Path]) -> bool: for lib in libraries: include_dirs += lib.include_dirs + unflags = _unflag_tokens() ( project_compile_flags, project_link_flags, project_lib_dirs, project_libs, - ) = _project_flags() + ) = _project_flags(unflags) defines = _defines_flags( config, esp8266_data[KEY_FLASH_MODE], board, board_build["defines"] ) @@ -526,7 +540,6 @@ def write_project(paths: dict[str, Path]) -> bool: # build_unflags applies to the framework flag sets too (compile and link), # as under PlatformIO (a silently ignored ``build_unflags: -Os`` would # diverge between the toolchains). - unflags = _unflag_tokens() cflags = [f for f in cflags if f not in unflags] cxxflags = [f for f in cxxflags if f not in unflags] asflags = [f for f in asflags if f not in unflags] diff --git a/tests/unit_tests/build_gen/test_arduino8266.py b/tests/unit_tests/build_gen/test_arduino8266.py index 180a5aadf3..351c054b0c 100644 --- a/tests/unit_tests/build_gen/test_arduino8266.py +++ b/tests/unit_tests/build_gen/test_arduino8266.py @@ -582,7 +582,9 @@ def test_project_flags_trailing_bare_linker_flag_warns( caplog: pytest.LogCaptureFixture, ) -> None: _set_flags("-l") - compile_flags, link_flags, lib_dirs, libs = arduino8266._project_flags() + compile_flags, link_flags, lib_dirs, libs = arduino8266._project_flags( + arduino8266._unflag_tokens() + ) assert "Ignoring trailing '-l'" in caplog.text assert not libs assert not lib_dirs @@ -592,7 +594,9 @@ def test_project_flags_trailing_bare_linker_flag_warns( def test_project_flags_lexed_entry_scatters_non_linker_tokens() -> None: _set_flags("-L /d -Wl,-Map=m stray") - compile_flags, link_flags, lib_dirs, libs = arduino8266._project_flags() + compile_flags, link_flags, lib_dirs, libs = arduino8266._project_flags( + arduino8266._unflag_tokens() + ) assert lib_dirs == [Path("/d")] assert link_flags == ["-Wl,-Map=m"] assert "stray" in compile_flags @@ -612,7 +616,9 @@ def test_flag_defines_lexes_multi_token_entries() -> None: def test_project_flags_lexes_every_entry() -> None: """A linker flag anywhere in an entry reaches the link line (PIO parity).""" _set_flags("-DFOO=1 -lbar") - compile_flags, _link, _dirs, libs = arduino8266._project_flags() + compile_flags, _link, _dirs, libs = arduino8266._project_flags( + arduino8266._unflag_tokens() + ) assert libs == ["bar"] assert "-DFOO=1" in compile_flags @@ -621,7 +627,9 @@ def test_project_flags_unflags_match_tokens() -> None: """build_unflags removes a token embedded in a multi-token entry.""" _set_flags("-Os -g3") CORE.build_unflags = {"-Os"} - compile_flags, _link, _dirs, _libs = arduino8266._project_flags() + compile_flags, _link, _dirs, _libs = arduino8266._project_flags( + arduino8266._unflag_tokens() + ) assert "-g3" in compile_flags assert "-Os" not in compile_flags @@ -629,7 +637,22 @@ def test_project_flags_unflags_match_tokens() -> None: def test_project_flags_requotes_lexed_defines() -> None: """A quoted spaced value stays one compiler argument after lex/emit.""" _set_flags('-DGREETING="hello world"') - compile_flags, _link, _dirs, _libs = arduino8266._project_flags() + compile_flags, _link, _dirs, _libs = arduino8266._project_flags( + arduino8266._unflag_tokens() + ) # shlex folds the quotes (as PIO's ParseFlags does); _shell_token # re-quotes the spaced token so the shell passes one argv element assert compile_flags == ['"-DGREETING=hello world"'] + + +def test_shell_token_escaping() -> None: + """Tokens survive both POSIX sh and the Windows CRT argv parser.""" + assert arduino8266._shell_token("-Os") == "-Os" + # $ would be expanded (to nothing) by ninja itself + assert arduino8266._shell_token("-DX=$HOME") == "-DX=$$HOME" + # Backslashes not before a quote stay single (Windows path in a define) + assert arduino8266._shell_token("-DP=C:\\x y") == '"-DP=C:\\x y"' + # A quote is escaped and the preceding backslash run doubles + assert arduino8266._shell_token('-DX=a\\"b c') == '"-DX=a\\\\\\"b c"' + # A trailing backslash run doubles before the closing quote + assert arduino8266._shell_token("a b\\") == '"a b\\\\"' diff --git a/tests/unit_tests/test_arduino8266_framework.py b/tests/unit_tests/test_arduino8266_framework.py index 7dcbfd0d32..df6c8cc423 100644 --- a/tests/unit_tests/test_arduino8266_framework.py +++ b/tests/unit_tests/test_arduino8266_framework.py @@ -2,6 +2,7 @@ from __future__ import annotations +from contextlib import contextmanager import os from pathlib import Path import subprocess @@ -306,6 +307,11 @@ def test_check_and_install_returns_paths(tmp_path: Path) -> None: ) assert paths["ninja_path"] == tmp_path / "ninja" assert mock_install.call_count == 2 + # The layout checks cover the directories write_project needs, including + # the bundled libraries/ tree + fw_expect = mock_install.call_args_list[0].kwargs["expect"] + assert set(fw_expect) == {"cores/esp8266", "tools/sdk", "libraries"} + assert mock_install.call_args_list[1].kwargs["expect"] == ("bin",) def test_get_build_env_prepends_toolchain_bin(tmp_path: Path) -> None: @@ -409,20 +415,14 @@ def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None: dest = tmp_path / "pkg" marker = dest / ".esphome_extracted" - class _FakeLock: - def __init__(self, *_a, **_kw) -> None: - pass - - def __enter__(self): - dest.mkdir(parents=True, exist_ok=True) - marker.touch() - return self - - def __exit__(self, *args) -> None: - pass + @contextmanager + def _fake_lock(*_a, **_kw): + dest.mkdir(parents=True, exist_ok=True) + marker.touch() + yield with ( - patch("filelock.FileLock", _FakeLock), + patch("filelock.FileLock", _fake_lock), patch.object(framework, "download_from_mirrors") as mock_download, patch.object(framework, "rmdir") as mock_rmdir, ):