diff --git a/plugins/google_meet/cli.py b/plugins/google_meet/cli.py index 1ab0c0fd73db..1d7c823211f1 100644 --- a/plugins/google_meet/cli.py +++ b/plugins/google_meet/cli.py @@ -293,7 +293,7 @@ def _confirm(prompt: str) -> bool: elif system == "Darwin": have_bh = False try: - out = _sp.check_output(["system_profiler", "SPAudioDataType"], text=True) + out = _sp.check_output(["system_profiler", "SPAudioDataType"], text=True) # windows-footgun: ok — macOS only (system_profiler), guarded by platform check above have_bh = "BlackHole" in out except Exception: pass diff --git a/scripts/check-windows-footguns.py b/scripts/check-windows-footguns.py index 7ae7ca50c4e7..ba73daca08d7 100644 --- a/scripts/check-windows-footguns.py +++ b/scripts/check-windows-footguns.py @@ -324,6 +324,55 @@ class Footgun: " pass # Windows asyncio doesn't support signal handlers" ), ), + Footgun( + name="subprocess text=True without explicit encoding=", + # Match ``text=True`` (or ``text = True``) anywhere on a line. We + # rely on the post_filter to (a) skip lines that already pass + # ``encoding=`` on the same line, and (b) skip false positives like + # ``def text(self, ...)`` or string literals. ``text=True`` is + # overwhelmingly a subprocess kwarg, so a bare match + filter has a + # high signal-to-noise ratio and avoids the complexity of parsing + # multi-line subprocess calls (which the line-based scanner can't + # reliably attribute to a single line anyway). + pattern=re.compile(r"\btext\s*=\s*True\b"), + message=( + "subprocess text=True without explicit encoding= decodes " + "child output with locale.getpreferredencoding() — cp936 " + "(GBK) on Chinese Windows, cp1252 on Western Windows — " + "which crashes _readerthread with UnicodeDecodeError on " + "non-default-codepage bytes. Always pass encoding='utf-8' " + "(and errors='replace' for Windows-native CLIs that emit " + "non-UTF-8). See issues #47939, #53428, #57238." + ), + fix=( + "subprocess.run(..., text=True, encoding='utf-8', " + "errors='replace')\n" + "Both params are required: encoding alone still crashes on " + "non-UTF-8 bytes from Windows-native CLIs (tasklist, " + "schtasks)." + ), + post_filter=lambda m, line: ( + # Skip if the same line already specifies encoding=. + "encoding=" not in line + and "encoding =" not in line + # Skip method definitions named ``text`` (def text(self, ...)). + and not line.lstrip().startswith("def ") + and not line.lstrip().startswith("async def ") + # Skip ``text=True`` inside string literals (heuristic: the + # substring appears between matching quotes that aren't part + # of an f-string expression). This is imperfect but catches + # the common case of docstrings mentioning text=True. + and not _looks_like_string_literal(line, m) + # Skip lines that are obviously not subprocess calls — e.g. + # DataFrame.rename(text=True) or similar. We can't know for + # sure without parsing, so we accept some false negatives by + # only flagging when ``subprocess`` or a known subprocess- + # shaped call (run/Popen/call/check_output/check_call/ + # check_output) appears on the same line. This keeps the + # rule focused on the actual footgun. + and _is_likely_subprocess_call(line) + ), + ), ] @@ -408,6 +457,66 @@ def _find_unquoted_hash(line: str) -> int | None: return None +# Subprocess method names that accept ``text=`` and are affected by the +# encoding-default footgun. Used by ``_is_likely_subprocess_call`` below to +# keep the ``text=True`` rule focused on subprocess calls (and avoid flagging +# unrelated APIs that happen to accept a ``text`` kwarg). +_SUBPROCESS_METHODS = ( + "subprocess.run", + "subprocess.Popen", + "subprocess.call", + "subprocess.check_output", + "subprocess.check_call", + "_sp.run", # common alias + "_sp.Popen", + "_sp.check_output", + "_sp.check_call", + "_sp.call", + ".run(", # bare .run( — usually subprocess.run + ".Popen(", + ".check_output(", + ".check_call(", + ".call(", +) + + +def _is_likely_subprocess_call(line: str) -> bool: + """Heuristic: does this line look like a subprocess invocation? + + The ``text=True`` footgun rule only fires when the matched line also + contains a subprocess-shaped call site. This avoids false positives on + unrelated APIs that accept a ``text`` kwarg (e.g. DataFrame.rename, + custom library calls). Multi-line calls where the ``subprocess.X(`` + prefix is on a previous line won't be flagged — that's an acceptable + false negative for a line-based scanner. + """ + return any(token in line for token in _SUBPROCESS_METHODS) + + +def _looks_like_string_literal(line: str, match: "re.Match") -> bool: + """Heuristic: is the ``text=True`` match inside a string literal? + + Catches the common case of docstrings/comments that mention ``text=True`` + as prose. Walks the line tracking single/double quote state and returns + True if the match start index falls inside a quoted region. + """ + start = match.start() + in_s = False + in_d = False + i = 0 + while i < start and i < len(line): + c = line[i] + if c == "\\" and (in_s or in_d) and i + 1 < len(line): + i += 2 + continue + if not in_d and c == "'": + in_s = not in_s + elif not in_s and c == '"': + in_d = not in_d + i += 1 + return in_s or in_d + + def scan_file(path: Path, footguns: list[Footgun]) -> list[tuple[int, str, Footgun]]: """Return a list of (line_number, line, footgun) for unsuppressed matches.""" try: @@ -493,6 +602,8 @@ def get_staged_files() -> list[Path]: cwd=REPO_ROOT, stderr=subprocess.DEVNULL, text=True, + encoding="utf-8", + errors="replace", ) except (subprocess.CalledProcessError, FileNotFoundError): return [] @@ -507,6 +618,8 @@ def get_diff_files(ref: str) -> list[Path]: cwd=REPO_ROOT, stderr=subprocess.DEVNULL, text=True, + encoding="utf-8", + errors="replace", ) except (subprocess.CalledProcessError, FileNotFoundError): return [] diff --git a/tests/scripts/test_footgun_subprocess_encoding.py b/tests/scripts/test_footgun_subprocess_encoding.py new file mode 100644 index 000000000000..00b29aec913a --- /dev/null +++ b/tests/scripts/test_footgun_subprocess_encoding.py @@ -0,0 +1,276 @@ +"""Tests for the ``subprocess text=True without explicit encoding=`` footgun +rule in ``scripts/check-windows-footguns.py``. + +This rule (added alongside PR #60741) catches ``subprocess.run/Popen/call/ +check_output/check_call(..., text=True, ...)`` calls that don't pass an +explicit ``encoding=``. On Chinese Windows (cp936/GBK) and other non-UTF-8 +default codepages, ``text=True`` without ``encoding=`` decodes child output +with ``locale.getpreferredencoding(False)`` and crashes ``_readerthread`` +with ``UnicodeDecodeError`` on non-default-codepage bytes. + +See issues #47939, #53428, #57238. +""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +LINTER_PATH = REPO_ROOT / "scripts" / "check-windows-footguns.py" + + +def _load_linter_module(): + """Import the linter script as a module (it's not a package). + + Register the module in sys.modules BEFORE exec_module so that + ``@dataclass`` can resolve ``cls.__module__`` via + ``sys.modules.get(cls.__module__).__dict__`` (CPython 3.11+ dataclass + internals require this). + """ + spec = importlib.util.spec_from_file_location("check_windows_footguns", LINTER_PATH) + mod = importlib.util.module_from_spec(spec) + sys.modules["check_windows_footguns"] = mod + spec.loader.exec_module(mod) + return mod + + +@pytest.fixture(scope="module") +def linter(): + return _load_linter_module() + + +def _find_footgun(linter, name: str): + """Locate a Footgun by name in the FOOTGUNS list.""" + for fg in linter.FOOTGUNS: + if fg.name == name: + return fg + pytest.fail(f"Footgun rule '{name}' not found in FOOTGUNS") + + +def _scan_line(linter, line: str, footgun_name: str) -> bool: + """Return True if the given line triggers the named footgun rule. + + Uses the linter's own pattern + post_filter logic so the test exercises + the real detection path (including guard-hint and suppression checks). + """ + fg = _find_footgun(linter, footgun_name) + # Replicate the relevant checks from scan_file(): suppression marker, + # guard hints, then pattern + post_filter. + if linter.SUPPRESS_MARKER.search(line): + return False + if any(hint in line for hint in linter.GUARD_HINTS): + return False + code = linter._strip_code(line) + if not code.strip(): + return False + match = fg.pattern.search(code) + if not match: + return False + if fg.post_filter is not None: + try: + if not fg.post_filter(match, line): + return False + except (IndexError, AttributeError): + return False + return True + + +RULE_NAME = "subprocess text=True without explicit encoding=" + + +# --------------------------------------------------------------------------- +# Detection — these SHOULD be flagged +# --------------------------------------------------------------------------- + + +class TestDetection: + def test_flags_subprocess_run_text_true_without_encoding(self, linter): + line = ' result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)' + assert _scan_line(linter, line, RULE_NAME), "expected flag for text=True without encoding=" + + def test_flags_subprocess_popen_text_true_without_encoding(self, linter): + line = ' p = subprocess.Popen(cmd, text=True, stdout=PIPE)' + assert _scan_line(linter, line, RULE_NAME) + + def test_flags_subprocess_check_output_text_true(self, linter): + line = ' out = subprocess.check_output(["git", "status"], text=True)' + assert _scan_line(linter, line, RULE_NAME) + + def test_flags_sp_alias_text_true(self, linter): + line = ' res = _sp.run(cmd, text=True, timeout=5)' + assert _scan_line(linter, line, RULE_NAME) + + def test_flags_text_with_spaces_around_equals(self, linter): + line = ' subprocess.run(cmd, text = True, timeout=10)' + assert _scan_line(linter, line, RULE_NAME) + + def test_flags_bare_run_call(self, linter): + # .run( without explicit subprocess. prefix — still a subprocess call + line = ' result = obj.run(cmd, text=True)' + assert _scan_line(linter, line, RULE_NAME) + + +# --------------------------------------------------------------------------- +# Suppression — these should NOT be flagged +# --------------------------------------------------------------------------- + + +class TestSuppression: + def test_does_not_flag_when_encoding_present(self, linter): + line = ' subprocess.run(cmd, text=True, encoding="utf-8", errors="replace")' + assert not _scan_line(linter, line, RULE_NAME) + + def test_does_not_flag_when_encoding_with_spaces(self, linter): + line = " subprocess.run(cmd, text=True, encoding = 'utf-8')" + assert not _scan_line(linter, line, RULE_NAME) + + def test_does_not_flag_inline_suppression_marker(self, linter): + line = ' subprocess.run(cmd, text=True) # windows-footgun: ok — POSIX only' + assert not _scan_line(linter, line, RULE_NAME) + + def test_does_not_flag_non_subprocess_text_kwarg(self, linter): + # DataFrame.rename(text=True) — not a subprocess call + line = ' df = df.rename(text=True)' + assert not _scan_line(linter, line, RULE_NAME), ( + "should not flag non-subprocess APIs that accept text= kwarg" + ) + + def test_does_not_flag_text_true_in_string_literal(self, linter): + line = ' """See subprocess.run(text=True) for details."""' + assert not _scan_line(linter, line, RULE_NAME), ( + "should not flag text=True inside docstrings" + ) + + def test_does_not_flag_def_text_method(self, linter): + line = ' def text(self, value: bool = True):' + assert not _scan_line(linter, line, RULE_NAME) + + def test_does_not_flag_comment_only_line(self, linter): + line = ' # subprocess.run(cmd, text=True) — example' + assert not _scan_line(linter, line, RULE_NAME) + + +# --------------------------------------------------------------------------- +# Helper functions — unit tests for _is_likely_subprocess_call and +# _looks_like_string_literal +# --------------------------------------------------------------------------- + + +class TestHelpers: + def test_is_likely_subprocess_call_matches_subprocess_run(self, linter): + assert linter._is_likely_subprocess_call("subprocess.run(cmd, text=True)") + + def test_is_likely_subprocess_call_matches_bare_run(self, linter): + assert linter._is_likely_subprocess_call("result = obj.run(cmd, text=True)") + + def test_is_likely_subprocess_call_rejects_dataframe(self, linter): + assert not linter._is_likely_subprocess_call("df.rename(text=True)") + + def test_is_likely_subprocess_call_rejects_plain_assignment(self, linter): + assert not linter._is_likely_subprocess_call("config.text = True") + + def test_looks_like_string_literal_double_quotes(self, linter): + import re + line = ' msg = "use text=True carefully"' + match = re.search(r"\btext\s*=\s*True\b", line) + assert match is not None + assert linter._looks_like_string_literal(line, match) + + def test_looks_like_string_literal_single_quotes(self, linter): + import re + line = " msg = 'see text=True in docs'" + match = re.search(r"\btext\s*=\s*True\b", line) + assert match is not None + assert linter._looks_like_string_literal(line, match) + + def test_looks_like_string_literal_false_for_real_code(self, linter): + import re + line = ' subprocess.run(cmd, text=True)' + match = re.search(r"\btext\s*=\s*True\b", line) + assert match is not None + assert not linter._looks_like_string_literal(line, match) + + +# --------------------------------------------------------------------------- +# Full-repo scan — after PR #60741 merges, the new rule should find ZERO +# unsuppressed violations in the whole tree (excluding the linter itself +# and CONTRIBUTING docs). This test will FAIL until PR #60741 is merged; +# mark it xfail when run on a branch that doesn't include PR #60741's fixes. +# --------------------------------------------------------------------------- + + +class TestFullRepoScan: + def test_new_rule_find_only_known_violations(self, linter, monkeypatch): + """Scan the full repo and assert the new rule's matches are exactly + the set of call sites that PR #60741 fixes (or zero, if PR #60741 + is already merged into this branch). + + This is a regression guard: if someone adds a new + ``subprocess.run(text=True)`` without ``encoding=``, this test + catches it. + """ + # The 7 call sites that PR #60741 fixes. If PR #60741 is merged + # into this branch, this set should be empty. If not, these are + # the expected matches. + pr_60741_sites = { + "hermes_cli/main.py", + "hermes_cli/onepassword_secrets_cli.py", + "hermes_cli/setup.py", + "tools/transcription_tools.py", + "tools/tts_tool.py", + } + + # Run the full scan + roots = [ + REPO_ROOT / "hermes_cli", + REPO_ROOT / "gateway", + REPO_ROOT / "tools", + REPO_ROOT / "cron", + REPO_ROOT / "agent", + REPO_ROOT / "plugins", + REPO_ROOT / "scripts", + REPO_ROOT / "acp_adapter", + REPO_ROOT / "acp_registry", + ] + roots = [r for r in roots if r.exists()] + + fg = _find_footgun(linter, RULE_NAME) + new_rule_matches: dict[str, list[int]] = {} + + for path in linter.iter_files(roots): + matches = linter.scan_file(path, [fg]) # scan with ONLY the new rule + if matches: + rel = path.relative_to(REPO_ROOT).as_posix() + new_rule_matches[rel] = [m[0] for m in matches] + + # Determine which sites remain. PR #60741's fixes are on a separate + # branch; if this branch doesn't include them, the 7 call sites + # will still be flagged — that's expected, not a failure. + if new_rule_matches: + # Filter out the linter itself (it mentions text=True in its + # own pattern/message, but EXCLUDED_FILES handles that for the + # CLI entry point; the helper functions could trip it). + new_rule_matches = { + k: v for k, v in new_rule_matches.items() + if k != "scripts/check-windows-footguns.py" + } + + if not new_rule_matches: + # PR #60741 already merged — clean tree. This is the goal state. + return + + # Matches remain — they must be exactly the PR #60741 sites. + matched_files = set(new_rule_matches.keys()) + unexpected = matched_files - pr_60741_sites + if unexpected: + pytest.fail( + f"New footgun rule found UNEXPECTED matches in files not " + f"covered by PR #60741: {sorted(unexpected)}.\n" + f"These are either new regressions or call sites that need " + f"a `# windows-footgun: ok` suppression." + ) + # All matches are the expected PR #60741 sites — OK on this branch. diff --git a/tools/environments/singularity.py b/tools/environments/singularity.py index 666d908b2568..44ffe7d727c4 100644 --- a/tools/environments/singularity.py +++ b/tools/environments/singularity.py @@ -220,7 +220,7 @@ def _start_instance(self): cmd.extend([str(self.image), self.instance_id]) try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL) # windows-footgun: ok — Singularity (Linux HPC) only, never runs on Windows if result.returncode != 0: raise RuntimeError(f"Failed to start instance: {result.stderr}") self._instance_started = True diff --git a/tools/voice_mode.py b/tools/voice_mode.py index d000e29d59d9..dc45cadf792e 100644 --- a/tools/voice_mode.py +++ b/tools/voice_mode.py @@ -389,7 +389,7 @@ def start(self, on_silence_stop=None) -> None: "-c", str(CHANNELS), ] try: - subprocess.run(command, capture_output=True, text=True, timeout=15, check=True, stdin=subprocess.DEVNULL) + subprocess.run(command, capture_output=True, text=True, timeout=15, check=True, stdin=subprocess.DEVNULL) # windows-footgun: ok — Termux (Android) only, never runs on Windows except subprocess.CalledProcessError as e: details = (e.stderr or e.stdout or str(e)).strip() raise RuntimeError(f"Termux microphone start failed: {details}") from e @@ -406,7 +406,7 @@ def _stop_termux_recording(self) -> None: mic_cmd = _termux_microphone_command() if not mic_cmd: return - subprocess.run([mic_cmd, "-q"], capture_output=True, text=True, timeout=15, check=False, stdin=subprocess.DEVNULL) + subprocess.run([mic_cmd, "-q"], capture_output=True, text=True, timeout=15, check=False, stdin=subprocess.DEVNULL) # windows-footgun: ok — Termux (Android) only, never runs on Windows def stop(self) -> Optional[str]: with self._lock: